README.md

@@ -110,6 +110,7 @@ Use the `addBodyPart` method to add a multipart part to the request.
 This part can be of type:
 * `ByteArrayPart`
 * `FilePart`
+* `InputStreamPart`
 * `StringPart`
 
 ### Dealing with Responses

client/pom.xml

@@ -2,7 +2,7 @@
   <parent>
     <groupId>org.asynchttpclient</groupId>
     <artifactId>async-http-client-project</artifactId>
-    <version>2.5.3</version>
+    <version>2.6.0</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
   <artifactId>async-http-client</artifactId>
@@ -73,5 +73,10 @@
       <artifactId>reactive-streams-examples</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.kerby</groupId>
+      <artifactId>kerb-simplekdc</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 </project>

client/src/main/java/org/asynchttpclient/Dsl.java

@@ -99,7 +99,11 @@ public final class Dsl {
             .setNtlmDomain(prototype.getNtlmDomain())
             .setNtlmHost(prototype.getNtlmHost())
             .setUseAbsoluteURI(prototype.isUseAbsoluteURI())
-            .setOmitQuery(prototype.isOmitQuery());
+            .setOmitQuery(prototype.isOmitQuery())
+            .setServicePrincipalName(prototype.getServicePrincipalName())
+            .setUseCanonicalHostname(prototype.isUseCanonicalHostname())
+            .setCustomLoginConfig(prototype.getCustomLoginConfig())
+            .setLoginContextName(prototype.getLoginContextName());
   }
 
   public static Realm.Builder realm(AuthScheme scheme, String principal, String password) {

client/src/main/java/org/asynchttpclient/Realm.java

@@ -23,6 +23,7 @@ import org.asynchttpclient.util.StringUtils;
 
 import java.nio.charset.Charset;
 import java.security.MessageDigest;
+import java.util.Map;
 import java.util.concurrent.ThreadLocalRandom;
 
 import static java.nio.charset.StandardCharsets.*;
@@ -60,6 +61,10 @@ public class Realm {
   private final String ntlmDomain;
   private final boolean useAbsoluteURI;
   private final boolean omitQuery;
+  private final Map<String, String> customLoginConfig;
+  private final String servicePrincipalName;
+  private final boolean useCanonicalHostname;
+  private final String loginContextName;
 
   private Realm(AuthScheme scheme,
                 String principal,
@@ -78,11 +83,15 @@ public class Realm {
                 String ntlmDomain,
                 String ntlmHost,
                 boolean useAbsoluteURI,
-                boolean omitQuery) {
+                boolean omitQuery,
+                String servicePrincipalName,
+                boolean useCanonicalHostname,
+                Map<String, String> customLoginConfig,
+                String loginContextName) {
 
     this.scheme = assertNotNull(scheme, "scheme");
-    this.principal = assertNotNull(principal, "principal");
-    this.password = assertNotNull(password, "password");
+    this.principal = principal;
+    this.password = password;
     this.realmName = realmName;
     this.nonce = nonce;
     this.algorithm = algorithm;
@@ -98,6 +107,10 @@ public class Realm {
     this.ntlmHost = ntlmHost;
     this.useAbsoluteURI = useAbsoluteURI;
     this.omitQuery = omitQuery;
+    this.servicePrincipalName = servicePrincipalName;
+    this.useCanonicalHostname = useCanonicalHostname;
+    this.customLoginConfig = customLoginConfig;
+    this.loginContextName = loginContextName;
   }
 
   public String getPrincipal() {
@@ -187,12 +200,48 @@ public class Realm {
     return omitQuery;
   }
 
+  public Map<String, String> getCustomLoginConfig() {
+    return customLoginConfig;
+  }
+
+  public String getServicePrincipalName() {
+    return servicePrincipalName;
+  }
+
+  public boolean isUseCanonicalHostname() {
+    return useCanonicalHostname;
+  }
+
+  public String getLoginContextName() {
+    return loginContextName;
+  }
+
   @Override
   public String toString() {
-    return "Realm{" + "principal='" + principal + '\'' + ", scheme=" + scheme + ", realmName='" + realmName + '\''
-            + ", nonce='" + nonce + '\'' + ", algorithm='" + algorithm + '\'' + ", response='" + response + '\''
-            + ", qop='" + qop + '\'' + ", nc='" + nc + '\'' + ", cnonce='" + cnonce + '\'' + ", uri='" + uri + '\''
-            + ", useAbsoluteURI='" + useAbsoluteURI + '\'' + ", omitQuery='" + omitQuery + '\'' + '}';
+    return "Realm{" +
+        "principal='" + principal + '\'' +
+        ", password='" + password + '\'' +
+        ", scheme=" + scheme +
+        ", realmName='" + realmName + '\'' +
+        ", nonce='" + nonce + '\'' +
+        ", algorithm='" + algorithm + '\'' +
+        ", response='" + response + '\'' +
+        ", opaque='" + opaque + '\'' +
+        ", qop='" + qop + '\'' +
+        ", nc='" + nc + '\'' +
+        ", cnonce='" + cnonce + '\'' +
+        ", uri=" + uri +
+        ", usePreemptiveAuth=" + usePreemptiveAuth +
+        ", charset=" + charset +
+        ", ntlmHost='" + ntlmHost + '\'' +
+        ", ntlmDomain='" + ntlmDomain + '\'' +
+        ", useAbsoluteURI=" + useAbsoluteURI +
+        ", omitQuery=" + omitQuery +
+        ", customLoginConfig=" + customLoginConfig +
+        ", servicePrincipalName='" + servicePrincipalName + '\'' +
+        ", useCanonicalHostname=" + useCanonicalHostname +
+        ", loginContextName='" + loginContextName + '\'' +
+        '}';
   }
 
   public enum AuthScheme {
@@ -223,6 +272,18 @@ public class Realm {
     private String ntlmHost = "localhost";
     private boolean useAbsoluteURI = false;
     private boolean omitQuery;
+    /**
+     * Kerberos/Spnego properties
+     */
+    private Map<String, String> customLoginConfig;
+    private String servicePrincipalName;
+    private boolean useCanonicalHostname;
+    private String loginContextName;
+
+    public Builder() {
+      this.principal = null;
+      this.password = null;
+    }
 
     public Builder(String principal, String password) {
       this.principal = principal;
@@ -311,6 +372,26 @@ public class Realm {
       return this;
     }
 
+    public Builder setCustomLoginConfig(Map<String, String> customLoginConfig) {
+      this.customLoginConfig = customLoginConfig;
+      return this;
+    }
+
+    public Builder setServicePrincipalName(String servicePrincipalName) {
+      this.servicePrincipalName = servicePrincipalName;
+      return this;
+    }
+
+    public Builder setUseCanonicalHostname(boolean useCanonicalHostname) {
+      this.useCanonicalHostname = useCanonicalHostname;
+      return this;
+    }
+
+    public Builder setLoginContextName(String loginContextName) {
+      this.loginContextName = loginContextName;
+      return this;
+    }
+
     private String parseRawQop(String rawQop) {
       String[] rawServerSupportedQops = rawQop.split(",");
       String[] serverSupportedQops = new String[rawServerSupportedQops.length];
@@ -501,7 +582,11 @@ public class Realm {
               ntlmDomain,
               ntlmHost,
               useAbsoluteURI,
-              omitQuery);
+              omitQuery,
+              servicePrincipalName,
+              useCanonicalHostname,
+              customLoginConfig,
+              loginContextName);
     }
   }
 }

client/src/main/java/org/asynchttpclient/RequestBuilderBase.java

@@ -267,7 +267,7 @@ public abstract class RequestBuilderBase<T extends RequestBuilderBase<T>> {
    * @param headers map of header names as the map keys and header values {@link Iterable} as the map values
    * @return {@code this}
    */
-  public T setHeaders(Map<CharSequence, ? extends Iterable<?>> headers) {
+  public T setHeaders(Map<? extends CharSequence, ? extends Iterable<?>> headers) {
     clearHeaders();
     if (headers != null) {
       headers.forEach((name, values) -> this.headers.add(name, values));
@@ -282,7 +282,7 @@ public abstract class RequestBuilderBase<T extends RequestBuilderBase<T>> {
    * @param headers map of header names as the map keys and header values as the map values
    * @return {@code this}
    */
-  public T setSingleHeaders(Map<CharSequence, ?> headers) {
+  public T setSingleHeaders(Map<? extends CharSequence, ?> headers) {
     clearHeaders();
     if (headers != null) {
       headers.forEach((name, value) -> this.headers.add(name, value));

client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java

@@ -345,7 +345,7 @@ public class ChannelManager {
       if (!isSslHandlerConfigured(pipeline)) {
         SslHandler sslHandler = createSslHandler(requestUri.getHost(), requestUri.getExplicitPort());
         whenHanshaked = sslHandler.handshakeFuture();
-        pipeline.addBefore(CHUNKED_WRITER_HANDLER, SSL_HANDLER, sslHandler);
+        pipeline.addBefore(INFLATER_HANDLER, SSL_HANDLER, sslHandler);
       }
       pipeline.addAfter(SSL_HANDLER, HTTP_CLIENT_CODEC, newHttpClientCodec());
 

client/src/main/java/org/asynchttpclient/netty/handler/AsyncHttpClientHandler.java

@@ -67,8 +67,10 @@ public abstract class AsyncHttpClientHandler extends ChannelInboundHandlerAdapte
     Object attribute = Channels.getAttribute(channel);
 
     try {
-      if (attribute instanceof OnLastHttpContentCallback && msg instanceof LastHttpContent) {
+      if (attribute instanceof OnLastHttpContentCallback) {
+        if (msg instanceof LastHttpContent) {
           ((OnLastHttpContentCallback) attribute).call();
+        }
 
       } else if (attribute instanceof NettyResponseFuture) {
         NettyResponseFuture<?> future = (NettyResponseFuture<?>) attribute;

client/src/main/java/org/asynchttpclient/netty/handler/intercept/ProxyUnauthorized407Interceptor.java

@@ -140,7 +140,7 @@ public class ProxyUnauthorized407Interceptor {
           return false;
         }
         try {
-          kerberosProxyChallenge(proxyServer, requestHeaders);
+          kerberosProxyChallenge(proxyRealm, proxyServer, requestHeaders);
 
         } catch (SpnegoEngineException e) {
           // FIXME
@@ -184,10 +184,17 @@ public class ProxyUnauthorized407Interceptor {
     return true;
   }
 
-  private void kerberosProxyChallenge(ProxyServer proxyServer,
+  private void kerberosProxyChallenge(Realm proxyRealm,
+                                      ProxyServer proxyServer,
                                       HttpHeaders headers) throws SpnegoEngineException {
 
-    String challengeHeader = SpnegoEngine.instance().generateToken(proxyServer.getHost());
+    String challengeHeader = SpnegoEngine.instance(proxyRealm.getPrincipal(),
+        proxyRealm.getPassword(),
+        proxyRealm.getServicePrincipalName(),
+        proxyRealm.getRealmName(),
+        proxyRealm.isUseCanonicalHostname(),
+        proxyRealm.getCustomLoginConfig(),
+        proxyRealm.getLoginContextName()).generateToken(proxyServer.getHost());
     headers.set(PROXY_AUTHORIZATION, NEGOTIATE + " " + challengeHeader);
   }
 

client/src/main/java/org/asynchttpclient/netty/handler/intercept/Unauthorized401Interceptor.java

@@ -139,7 +139,7 @@ public class Unauthorized401Interceptor {
           return false;
         }
         try {
-          kerberosChallenge(request, requestHeaders);
+          kerberosChallenge(realm, request, requestHeaders);
 
         } catch (SpnegoEngineException e) {
           // FIXME
@@ -200,12 +200,19 @@ public class Unauthorized401Interceptor {
     }
   }
 
-  private void kerberosChallenge(Request request,
+  private void kerberosChallenge(Realm realm,
+                                 Request request,
                                  HttpHeaders headers) throws SpnegoEngineException {
 
     Uri uri = request.getUri();
     String host = withDefault(request.getVirtualHost(), uri.getHost());
-    String challengeHeader = SpnegoEngine.instance().generateToken(host);
+    String challengeHeader = SpnegoEngine.instance(realm.getPrincipal(),
+        realm.getPassword(),
+        realm.getServicePrincipalName(),
+        realm.getRealmName(),
+        realm.isUseCanonicalHostname(),
+        realm.getCustomLoginConfig(),
+        realm.getLoginContextName()).generateToken(host);
     headers.set(AUTHORIZATION, NEGOTIATE + " " + challengeHeader);
   }
 }

client/src/main/java/org/asynchttpclient/netty/request/body/NettyBodyBody.java

@@ -53,7 +53,7 @@ public class NettyBodyBody implements NettyBody {
   public void write(final Channel channel, NettyResponseFuture<?> future) {
 
     Object msg;
-    if (body instanceof RandomAccessBody && !ChannelManager.isSslHandlerConfigured(channel.pipeline()) && !config.isDisableZeroCopy()) {
+    if (body instanceof RandomAccessBody && !ChannelManager.isSslHandlerConfigured(channel.pipeline()) && !config.isDisableZeroCopy() && getContentLength() > 0) {
       msg = new BodyFileRegion((RandomAccessBody) body);
 
     } else {

client/src/main/java/org/asynchttpclient/request/body/multipart/InputStreamPart.java

+/*
+ * Copyright (c) 2018 AsyncHttpClient Project. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at
+ *     http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+package org.asynchttpclient.request.body.multipart;
+
+import java.io.InputStream;
+import java.nio.charset.Charset;
+
+import static org.asynchttpclient.util.Assertions.assertNotNull;
+
+public class InputStreamPart extends FileLikePart {
+
+  private final InputStream inputStream;
+  private final long contentLength;
+
+  public InputStreamPart(String name, InputStream inputStream, String fileName) {
+    this(name, inputStream, fileName, -1);
+  }
+
+  public InputStreamPart(String name, InputStream inputStream, String fileName, long contentLength) {
+    this(name, inputStream, fileName, contentLength, null);
+  }
+
+  public InputStreamPart(String name, InputStream inputStream, String fileName, long contentLength, String contentType) {
+    this(name, inputStream, fileName, contentLength, contentType, null);
+  }
+
+  public InputStreamPart(String name, InputStream inputStream, String fileName, long contentLength, String contentType, Charset charset) {
+    this(name, inputStream, fileName, contentLength, contentType, charset, null);
+  }
+
+  public InputStreamPart(String name, InputStream inputStream, String fileName, long contentLength, String contentType, Charset charset,
+                         String contentId) {
+    this(name, inputStream, fileName, contentLength, contentType, charset, contentId, null);
+  }
+
+  public InputStreamPart(String name, InputStream inputStream, String fileName, long contentLength, String contentType, Charset charset,
+                         String contentId, String transferEncoding) {
+    super(name,
+            contentType,
+            charset,
+            fileName,
+            contentId,
+            transferEncoding);
+    this.inputStream = assertNotNull(inputStream, "inputStream");
+    this.contentLength = contentLength;
+  }
+
+  public InputStream getInputStream() {
+    return inputStream;
+  }
+
+  public long getContentLength() {
+    return contentLength;
+  }
+}

client/src/main/java/org/asynchttpclient/request/body/multipart/MultipartUtils.java

@@ -75,6 +75,9 @@ public class MultipartUtils {
       } else if (part instanceof StringPart) {
         multipartParts.add(new StringMultipartPart((StringPart) part, boundary));
 
+      } else if (part instanceof InputStreamPart) {
+        multipartParts.add(new InputStreamMultipartPart((InputStreamPart) part, boundary));
+
       } else {
         throw new IllegalArgumentException("Unknown part type: " + part);
       }

client/src/main/java/org/asynchttpclient/request/body/multipart/part/InputStreamMultipartPart.java

+/*
+ * Copyright (c) 2018 AsyncHttpClient Project. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at
+ *     http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+package org.asynchttpclient.request.body.multipart.part;
+
+import io.netty.buffer.ByteBuf;
+import org.asynchttpclient.netty.request.body.BodyChunkedInput;
+import org.asynchttpclient.request.body.multipart.InputStreamPart;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.WritableByteChannel;
+
+import static org.asynchttpclient.util.MiscUtils.closeSilently;
+
+public class InputStreamMultipartPart extends FileLikeMultipartPart<InputStreamPart> {
+
+  private long position = 0L;
+  private ByteBuffer buffer;
+  private ReadableByteChannel channel;
+
+  public InputStreamMultipartPart(InputStreamPart part, byte[] boundary) {
+    super(part, boundary);
+  }
+
+  private ByteBuffer getBuffer() {
+    if (buffer == null) {
+      buffer = ByteBuffer.allocateDirect(BodyChunkedInput.DEFAULT_CHUNK_SIZE);
+    }
+    return buffer;
+  }
+
+  private ReadableByteChannel getChannel() {
+    if (channel == null) {
+      channel = Channels.newChannel(part.getInputStream());
+    }
+    return channel;
+  }
+
+  @Override
+  protected long getContentLength() {
+    return part.getContentLength();
+  }
+
+  @Override
+  protected long transferContentTo(ByteBuf target) throws IOException {
+    InputStream inputStream = part.getInputStream();
+    int transferred = target.writeBytes(inputStream, target.writableBytes());
+    if (transferred > 0) {
+      position += transferred;
+    }
+    if (position == getContentLength() || transferred < 0) {
+      state = MultipartState.POST_CONTENT;
+      inputStream.close();
+    }
+    return transferred;
+  }
+
+  @Override
+  protected long transferContentTo(WritableByteChannel target) throws IOException {
+    ReadableByteChannel channel = getChannel();
+    ByteBuffer buffer = getBuffer();
+
+    int transferred = 0;
+    int read = channel.read(buffer);
+
+    if (read > 0) {
+      buffer.flip();
+      while (buffer.hasRemaining()) {
+        transferred += target.write(buffer);
+      }
+      buffer.compact();
+      position += transferred;
+    }
+    if (position == getContentLength() || read < 0) {
+      state = MultipartState.POST_CONTENT;
+      if (channel.isOpen()) {
+        channel.close();
+      }
+    }
+
+    return transferred;
+  }
+
+  @Override
+  public void close() {
+    super.close();
+    closeSilently(part.getInputStream());
+    closeSilently(channel);
+  }
+
+}

client/src/main/java/org/asynchttpclient/request/body/multipart/part/MultipartPart.java

@@ -106,6 +106,10 @@ public abstract class MultipartPart<T extends PartBase> implements Closeable {
   }
 
   public long length() {
+    long contentLength = getContentLength();
+    if (contentLength < 0) {
+      return contentLength;
+    }
     return preContentLength + postContentLength + getContentLength();
   }
 

client/src/main/java/org/asynchttpclient/spnego/NamePasswordCallbackHandler.java

+package org.asynchttpclient.spnego;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.PasswordCallback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import java.io.IOException;
+import java.lang.reflect.Method;
+
+public class NamePasswordCallbackHandler implements CallbackHandler {
+  private final Logger log = LoggerFactory.getLogger(getClass());
+  private static final String PASSWORD_CALLBACK_NAME = "setObject";
+  private static final Class<?>[] PASSWORD_CALLBACK_TYPES =
+      new Class<?>[] {Object.class, char[].class, String.class};
+
+  private String username;
+  private String password;
+
+  private String passwordCallbackName;
+
+  public NamePasswordCallbackHandler(String username, String password) {
+    this(username, password, null);
+  }
+
+  public NamePasswordCallbackHandler(String username, String password, String passwordCallbackName) {
+    this.username = username;
+    this.password = password;
+    this.passwordCallbackName = passwordCallbackName;
+  }
+
+  public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
+    for (int i = 0; i < callbacks.length; i++) {
+      Callback callback = callbacks[i];
+      if (handleCallback(callback)) {
+        continue;
+      } else if (callback instanceof NameCallback) {
+        ((NameCallback) callback).setName(username);
+      } else if (callback instanceof PasswordCallback) {
+        PasswordCallback pwCallback = (PasswordCallback) callback;
+        pwCallback.setPassword(password.toCharArray());
+      } else if (!invokePasswordCallback(callback)) {
+        String errorMsg = "Unsupported callback type " + callbacks[i].getClass().getName();
+        log.info(errorMsg);
+        throw new UnsupportedCallbackException(callbacks[i], errorMsg);
+      }
+    }
+  }
+
+  protected boolean handleCallback(Callback callback) {
+    return false;
+  }
+
+  /*
+   * This method is called from the handle(Callback[]) method when the specified callback
+   * did not match any of the known callback classes. It looks for the callback method
+   * having the specified method name with one of the suppported parameter types.
+   * If found, it invokes the callback method on the object and returns true.
+   * If not, it returns false.
+   */
+  private boolean invokePasswordCallback(Callback callback) {
+    String cbname = passwordCallbackName == null
+        ? PASSWORD_CALLBACK_NAME : passwordCallbackName;
+    for (Class<?> arg : PASSWORD_CALLBACK_TYPES) {
+      try {
+        Method method = callback.getClass().getMethod(cbname, arg);
+        Object args[] = new Object[] {
+            arg == String.class ? password : password.toCharArray()
+        };
+        method.invoke(callback, args);
+        return true;
+      } catch (Exception e) {
+        // ignore and continue
+        log.debug(e.toString());
+      }
+    }
+    return false;
+  }
+}
\ No newline at end of file

client/src/main/java/org/asynchttpclient/spnego/SpnegoEngine.java

@@ -38,6 +38,7 @@
 package org.asynchttpclient.spnego;
 
 import org.ietf.jgss.GSSContext;
+import org.ietf.jgss.GSSCredential;
 import org.ietf.jgss.GSSException;
 import org.ietf.jgss.GSSManager;
 import org.ietf.jgss.GSSName;
@@ -45,8 +46,19 @@ import org.ietf.jgss.Oid;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.security.auth.Subject;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.login.AppConfigurationEntry;
+import javax.security.auth.login.Configuration;
+import javax.security.auth.login.LoginContext;
+import javax.security.auth.login.LoginException;
 import java.io.IOException;
+import java.net.InetAddress;
+import java.security.PrivilegedActionException;
+import java.security.PrivilegedExceptionAction;
 import java.util.Base64;
+import java.util.HashMap;
+import java.util.Map;
 
 /**
  * SPNEGO (Simple and Protected GSSAPI Negotiation Mechanism) authentication scheme.
@@ -57,31 +69,87 @@ public class SpnegoEngine {
 
   private static final String SPNEGO_OID = "1.3.6.1.5.5.2";
   private static final String KERBEROS_OID = "1.2.840.113554.1.2.2";
-  private static SpnegoEngine instance;
+  private static Map<String, SpnegoEngine> instances = new HashMap<>();
   private final Logger log = LoggerFactory.getLogger(getClass());
   private final SpnegoTokenGenerator spnegoGenerator;
+  private final String username;
+  private final String password;
+  private final String servicePrincipalName;
+  private final String realmName;
+  private final boolean useCanonicalHostname;
+  private final String loginContextName;
+  private final Map<String, String> customLoginConfig;
 
-  public SpnegoEngine(final SpnegoTokenGenerator spnegoGenerator) {
+  public SpnegoEngine(final String username,
+                      final String password,
+                      final String servicePrincipalName,
+                      final String realmName,
+                      final boolean useCanonicalHostname,
+                      final Map<String, String> customLoginConfig,
+                      final String loginContextName,
+                      final SpnegoTokenGenerator spnegoGenerator) {
+    this.username = username;
+    this.password = password;
+    this.servicePrincipalName = servicePrincipalName;
+    this.realmName = realmName;
+    this.useCanonicalHostname = useCanonicalHostname;
+    this.customLoginConfig = customLoginConfig;
     this.spnegoGenerator = spnegoGenerator;
+    this.loginContextName = loginContextName;
   }
 
   public SpnegoEngine() {
-    this(null);
+    this(null,
+        null,
+        null,
+        null,
+        true,
+        null,
+        null,
+        null);
   }
 
-  public static SpnegoEngine instance() {
-    if (instance == null)
-      instance = new SpnegoEngine();
-    return instance;
+  public static SpnegoEngine instance(final String username,
+                                      final String password,
+                                      final String servicePrincipalName,
+                                      final String realmName,
+                                      final boolean useCanonicalHostname,
+                                      final Map<String, String> customLoginConfig,
+                                      final String loginContextName) {
+    String key = "";
+    if (customLoginConfig != null && !customLoginConfig.isEmpty()) {
+      StringBuilder customLoginConfigKeyValues = new StringBuilder();
+      for (String loginConfigKey : customLoginConfig.keySet()) {
+        customLoginConfigKeyValues.append(loginConfigKey).append("=")
+          .append(customLoginConfig.get(loginConfigKey));
+      }
+      key = customLoginConfigKeyValues.toString();
+    }
+    if (username != null) {
+      key += username;
+    }
+    if (loginContextName != null) {
+      key += loginContextName;
+    }
+    if (!instances.containsKey(key)) {
+      instances.put(key, new SpnegoEngine(username,
+          password,
+          servicePrincipalName,
+          realmName,
+          useCanonicalHostname,
+          customLoginConfig,
+          loginContextName,
+          null));
+    }
+    return instances.get(key);
   }
 
-  public String generateToken(String server) throws SpnegoEngineException {
+  public String generateToken(String host) throws SpnegoEngineException {
     GSSContext gssContext = null;
     byte[] token = null; // base64 decoded challenge
     Oid negotiationOid;
 
     try {
-      log.debug("init {}", server);
       /*
        * Using the SPNEGO OID is the correct method. Kerberos v5 works for IIS but not JBoss. Unwrapping the initial token when using SPNEGO OID looks like what is described
        * here...
@@ -99,10 +167,29 @@ public class SpnegoEngine {
       negotiationOid = new Oid(SPNEGO_OID);
 
       boolean tryKerberos = false;
+      String spn = getCompleteServicePrincipalName(host);
       try {
         GSSManager manager = GSSManager.getInstance();
-        GSSName serverName = manager.createName("HTTP@" + server, GSSName.NT_HOSTBASED_SERVICE);
-        gssContext = manager.createContext(serverName.canonicalize(negotiationOid), negotiationOid, null,
+        GSSName serverName = manager.createName(spn, GSSName.NT_HOSTBASED_SERVICE);
+        GSSCredential myCred = null;
+        if (username != null || loginContextName != null || (customLoginConfig != null && !customLoginConfig.isEmpty())) {
+          String contextName = loginContextName;
+          if (contextName == null) {
+            contextName = "";
+          }
+          LoginContext loginContext = new LoginContext(contextName,
+              null,
+              getUsernamePasswordHandler(),
+              getLoginConfiguration());
+          loginContext.login();
+          final Oid negotiationOidFinal = negotiationOid;
+          final PrivilegedExceptionAction<GSSCredential> action = () -> manager.createCredential(null,
+            GSSCredential.INDEFINITE_LIFETIME, negotiationOidFinal, GSSCredential.INITIATE_AND_ACCEPT);
+          myCred = Subject.doAs(loginContext.getSubject(), action);
+        }
+        gssContext = manager.createContext(useCanonicalHostname ? serverName.canonicalize(negotiationOid) : serverName,
+            negotiationOid,
+            myCred,
             GSSContext.DEFAULT_LIFETIME);
         gssContext.requestMutualAuth(true);
         gssContext.requestCredDeleg(true);
@@ -123,7 +210,7 @@ public class SpnegoEngine {
         log.debug("Using Kerberos MECH {}", KERBEROS_OID);
         negotiationOid = new Oid(KERBEROS_OID);
         GSSManager manager = GSSManager.getInstance();
-        GSSName serverName = manager.createName("HTTP@" + server, GSSName.NT_HOSTBASED_SERVICE);
+        GSSName serverName = manager.createName(spn, GSSName.NT_HOSTBASED_SERVICE);
         gssContext = manager.createContext(serverName.canonicalize(negotiationOid), negotiationOid, null,
                 GSSContext.DEFAULT_LIFETIME);
         gssContext.requestMutualAuth(true);
@@ -164,8 +251,59 @@ public class SpnegoEngine {
         throw new SpnegoEngineException(gsse.getMessage(), gsse);
       // other error
       throw new SpnegoEngineException(gsse.getMessage());
-    } catch (IOException ex) {
+    } catch (IOException | LoginException | PrivilegedActionException ex) {
       throw new SpnegoEngineException(ex.getMessage());
     }
   }
+
+  String getCompleteServicePrincipalName(String host) {
+    String name;
+    if (servicePrincipalName == null) {
+      if (useCanonicalHostname) {
+        host = getCanonicalHostname(host);
+      }
+      name = "HTTP@" + host;
+    } else {
+      name = servicePrincipalName;
+      if (realmName != null && !name.contains("@")) {
+        name += "@" + realmName;
+      }
+    }
+    log.debug("Service Principal Name is {}", name);
+    return name;
+  }
+
+  private String getCanonicalHostname(String hostname) {
+    String canonicalHostname = hostname;
+    try {
+      InetAddress in = InetAddress.getByName(hostname);
+      canonicalHostname = in.getCanonicalHostName();
+      log.debug("Resolved hostname={} to canonicalHostname={}", hostname, canonicalHostname);
+    } catch (Exception e) {
+      log.warn("Unable to resolve canonical hostname", e);
+    }
+    return canonicalHostname;
+  }
+
+  private CallbackHandler getUsernamePasswordHandler() {
+    if (username == null) {
+      return null;
+    }
+    return new NamePasswordCallbackHandler(username, password);
+  }
+
+  public Configuration getLoginConfiguration() {
+    if (customLoginConfig != null && !customLoginConfig.isEmpty()) {
+      return new Configuration() {
+        @Override
+        public AppConfigurationEntry[] getAppConfigurationEntry(String name) {
+          return new AppConfigurationEntry[] {
+              new AppConfigurationEntry("com.sun.security.auth.module.Krb5LoginModule",
+                  AppConfigurationEntry.LoginModuleControlFlag.REQUIRED,
+                  customLoginConfig)};
+        }
+      };
+    }
+    return null;
+  }
 }

client/src/main/java/org/asynchttpclient/util/AuthenticatorUtils.java

@@ -175,7 +175,14 @@ public final class AuthenticatorUtils {
             host = request.getUri().getHost();
 
           try {
-            authorizationHeader = NEGOTIATE + " " + SpnegoEngine.instance().generateToken(host);
+            authorizationHeader = NEGOTIATE + " " + SpnegoEngine.instance(
+                realm.getPrincipal(),
+                realm.getPassword(),
+                realm.getServicePrincipalName(),
+                realm.getRealmName(),
+                realm.isUseCanonicalHostname(),
+                realm.getCustomLoginConfig(),
+                realm.getLoginContextName()).generateToken(host);
           } catch (SpnegoEngineException e) {
             throw new RuntimeException(e);
           }

client/src/test/java/org/asynchttpclient/RequestBuilderTest.java

@@ -20,10 +20,7 @@ import io.netty.handler.codec.http.cookie.Cookie;
 import io.netty.handler.codec.http.cookie.DefaultCookie;
 import org.testng.annotations.Test;
 
-import java.io.IOException;
-import java.io.UnsupportedEncodingException;
 import java.util.*;
-import java.util.concurrent.ExecutionException;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.Collections.singletonList;
@@ -174,4 +171,16 @@ public class RequestBuilderTest {
     Request request = requestBuilder.build();
     assertEquals(request.getUrl(), "http://localhost?key=value");
   }
+
+  @Test
+  public void testSettingHeadersUsingMapWithStringKeys() {
+    Map<String, List<String>> headers = new HashMap<>();
+    headers.put("X-Forwarded-For", singletonList("10.0.0.1"));
+
+    RequestBuilder requestBuilder = new RequestBuilder();
+    requestBuilder.setHeaders(headers);
+    requestBuilder.setUrl("http://localhost");
+    Request request =  requestBuilder.build();
+    assertEquals(request.getHeaders().get("X-Forwarded-For"), "10.0.0.1");
+  }
 }

client/src/test/java/org/asynchttpclient/proxy/HttpsProxyTest.java

@@ -99,6 +99,23 @@ public class HttpsProxyTest extends AbstractBasicTest {
     }
   }
 
+  @Test
+  public void testDecompressBodyWithProxy() throws Exception {
+    AsyncHttpClientConfig config = config()
+      .setFollowRedirect(true)
+      .setProxyServer(proxyServer("localhost", port1).build())
+      .setUseInsecureTrustManager(true)
+      .build();
+    try (AsyncHttpClient asyncHttpClient = asyncHttpClient(config)) {
+      String body = "hello world";
+      Response r = asyncHttpClient.executeRequest(post(getTargetUrl2())
+        .setHeader("X-COMPRESS", "true")
+        .setBody(body)).get();
+      assertEquals(r.getStatusCode(), 200);
+      assertEquals(r.getResponseBody(), body);
+    }
+  }
+
   @Test
   public void testPooledConnectionsWithProxy() throws Exception {
 

client/src/test/java/org/asynchttpclient/request/body/InputStreamPartLargeFileTest.java

+/*
+ * Copyright (c) 2018 AsyncHttpClient Project. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at
+ *     http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+package org.asynchttpclient.request.body;
+
+import org.asynchttpclient.AbstractBasicTest;
+import org.asynchttpclient.AsyncHttpClient;
+import org.asynchttpclient.Response;
+import org.asynchttpclient.request.body.multipart.InputStreamPart;
+import org.eclipse.jetty.server.Request;
+import org.eclipse.jetty.server.handler.AbstractHandler;
+import org.testng.annotations.Test;
+
+import javax.servlet.ServletInputStream;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.*;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.asynchttpclient.Dsl.asyncHttpClient;
+import static org.asynchttpclient.Dsl.config;
+import static org.asynchttpclient.test.TestUtils.LARGE_IMAGE_FILE;
+import static org.asynchttpclient.test.TestUtils.createTempFile;
+import static org.testng.Assert.assertEquals;
+
+public class InputStreamPartLargeFileTest extends AbstractBasicTest {
+
+  @Override
+  public AbstractHandler configureHandler() throws Exception {
+    return new AbstractHandler() {
+
+      public void handle(String target, Request baseRequest, HttpServletRequest req, HttpServletResponse resp) throws IOException {
+
+        ServletInputStream in = req.getInputStream();
+        byte[] b = new byte[8192];
+
+        int count;
+        int total = 0;
+        while ((count = in.read(b)) != -1) {
+          b = new byte[8192];
+          total += count;
+        }
+        resp.setStatus(200);
+        resp.addHeader("X-TRANSFERRED", String.valueOf(total));
+        resp.getOutputStream().flush();
+        resp.getOutputStream().close();
+
+        baseRequest.setHandled(true);
+      }
+    };
+  }
+
+  @Test
+  public void testPutImageFile() throws Exception {
+    try (AsyncHttpClient client = asyncHttpClient(config().setRequestTimeout(100 * 6000))) {
+      InputStream inputStream = new BufferedInputStream(new FileInputStream(LARGE_IMAGE_FILE));
+      Response response = client.preparePut(getTargetUrl()).addBodyPart(new InputStreamPart("test", inputStream, LARGE_IMAGE_FILE.getName(), LARGE_IMAGE_FILE.length(), "application/octet-stream", UTF_8)).execute().get();
+      assertEquals(response.getStatusCode(), 200);
+    }
+  }
+
+  @Test
+  public void testPutImageFileUnknownSize() throws Exception {
+    try (AsyncHttpClient client = asyncHttpClient(config().setRequestTimeout(100 * 6000))) {
+      InputStream inputStream = new BufferedInputStream(new FileInputStream(LARGE_IMAGE_FILE));
+      Response response = client.preparePut(getTargetUrl()).addBodyPart(new InputStreamPart("test", inputStream, LARGE_IMAGE_FILE.getName(), -1, "application/octet-stream", UTF_8)).execute().get();
+      assertEquals(response.getStatusCode(), 200);
+    }
+  }
+
+  @Test
+  public void testPutLargeTextFile() throws Exception {
+    File file = createTempFile(1024 * 1024);
+    InputStream inputStream = new BufferedInputStream(new FileInputStream(file));
+
+    try (AsyncHttpClient client = asyncHttpClient(config().setRequestTimeout(100 * 6000))) {
+      Response response = client.preparePut(getTargetUrl())
+              .addBodyPart(new InputStreamPart("test", inputStream, file.getName(), file.length(), "application/octet-stream", UTF_8)).execute().get();
+      assertEquals(response.getStatusCode(), 200);
+    }
+  }
+
+  @Test
+  public void testPutLargeTextFileUnknownSize() throws Exception {
+    File file = createTempFile(1024 * 1024);
+    InputStream inputStream = new BufferedInputStream(new FileInputStream(file));
+
+    try (AsyncHttpClient client = asyncHttpClient(config().setRequestTimeout(100 * 6000))) {
+      Response response = client.preparePut(getTargetUrl())
+              .addBodyPart(new InputStreamPart("test", inputStream, file.getName(), -1, "application/octet-stream", UTF_8)).execute().get();
+      assertEquals(response.getStatusCode(), 200);
+    }
+  }
+}