diff --git a/.travis.yml b/.travis.yml index 4f7e1fcd9c..bd4de63cd3 100644 --- a/.travis.yml +++ b/.travis.yml @@ -2,26 +2,22 @@ dist: xenial language: minimal sudo: true services: - - docker - +- docker env: global: - - SERVICE_NAME=async-http-client - - REVISION=2.5.3-SNAPSHOT - - secure: vto8UyM7tMHE4tUwxmYC150orDzrz/6b33m8V6+yh1uHWJmtqhyshsjLKFMmiqkAHz/7jmrgMMlHcnYJ27IyNJsqd4WNQfUjf9nRuizbRKD+yhy7R+IKbgy8bSYizyvQ/4T07sJkOcebzXJigTlOjxpwDfKgBj7vqr5paXrir0Srs0jjgyMxnASFYaBvy2i6wPyPqslkbwHwTapWg3porl/EmSj9rmgHVnPED5CPYraMp+zTHJCpP6xgGn1N3wS+x3sUDZ4MSHeZLXbaSzmIIUBNbYSUjOnEspD0Pi/78G1cl6sW7KiSIAt+H6hoy1+rU6J4RyB8M6pSvNJDlgnYfemG/6onTAXA5n/0+dcs1X2skA3GscOmPoEUuPYyKeQwk2eXK67UC7LNSu1HSAeIxPgwKv4axmFPv/2GJtM70HP28cR35gzyWMMhsxGT7tuKE7W83bEuT/nUzXCWCC80LCOpRRHvfxczUO7bFDE9Be4K/xKtIAJiHZimcvixRoxRojPVSr8DrmeSFA+oGgE6F0hW6cQDynqF7FlmIb+6h3DVPWegQZqi7RP3UHy2qSwj2gAYAydd6FHib7juc8K4xIDdgZ53reCmQL6ULEwKF2unWDxLZ3urqDmhIc3qRH6it9VWdKoNr25EDgKXMVfiGoeKeZBVAl0KtdD6kDgLE9Q= - - secure: mLmKxZKFaaa98pVHMMcF/XEoXIebrOX7B9ZSLdGeB/2N9PWQnTRRThLrbd2gY1mpxPW+Nu6Nn4M0eh0g0kfYpypaetq7RO8GtNgRDW7qNDTr3q+dcV0+xW2YC+sAhvOnTMu8uz0bHf1y6iQMMvjggU9pyBD9iGtcANn5WXg+UW9cnlvx44MIzpKXt9RzItNpRgTC2/z+j8QL5zjwpMB2xT6EuP7ZFj1xnCQTAZ04qdOmM9Ro7rpvw0zYUIsYuEiyZMIgaGAsaGr00+VpbM9yRQTx7gGjvU6DLs3NfjJrl9ZyaR/vofAGjeV8DBNWBJwL6faq3y091joLIsWv4+tDfh+SZVwbUUhhHZUS0tuqebC0XAQMR5SAfOcO7t7blu9zk3esaeGYWaahQ3Jy1XupvuRQRu9Uj2p5tQTNRrDN9/8l4a9lwdhruDeHUeAwT1UGyx68WhXppWHuHw/Gbt20mo6nspuFPVJ5JrqdYrcfL4dUs+XY2sg4esweivOfrUy7F0noA85/Ol14P2wHyVp4aO+Bc8R+g87MxVlPm+vyW2YJQOi9EYFnPc6YyCu/RbXjvGh6y8izAN6BojeRJ9OJZjRZCG2s9CgL7a75ZOJEQ/ADpnJyhjD6XPlq7ZLwrh0j9fKLTYpk/zh26Ws/ZtGi0sMlwIXotGzVrIQUGgOmS00= - - secure: DPh5/VR6H3TSzReXgN6HVUoS7nvBX6nchs4esS0jLc/H34wAhC87d8g/8rn7GOvlf/jfr3ujUdl9EFGEy5WjfhNJYDqD/lLz2FmeNh8mI5ZX6mAhmd1CI7ow3miT/Dj/jdY/felWjgWRQIZRaYAv+qwoT96JSSEIO4lpfzafp4lXeD9oKp5t2Uk+Afz41x+e/OCoOW1VtUZi9VEiKbxcQCrwZRuHptTveJCi86zPeIokuUT+BBs0qQur960bq0c+XpkOC6tg3SLIQJJ/OSariiGfuap/A8ZObn9YFprmUYUUV9X8sOPuCBMOMQkqeG9DxNGEloHa9Xi+QpLTfFuLPvyp+PhZVYlMLvsH2GeWLNimlBqENaeKt/2DcO6pwf5TB0JcpV5tEPcjI3QVuqUEk6oXpwwAdoff4bN6Uafo3vKyQXEAsr26b590kxAhKxpI7nPmYtvHSkCfVxLh5Q2R6C0dpaGdamBRKjD+Lku2PJXE+lErk5SnK2BgBkiuuBaxVvUYO9RatucVVne+VDTJY6brcXJEEDs/myfK6lpF2cb5BM14l58iHWBQHmiDhGyl3CKRVXpavG9IuuVMf7siV0957ixCZstlU7J4RDwunGrl6RM3BQdj55atnHu+AKuy3piVy+NX+IimiaRsWw+N04pO3sAakG0fMa1YQ55RDOQ= - - secure: eZclfJ0N5XzH4Y8ZzLrbcX6iv9Mgc4gctU//jNTN39U2qlBvSUiuY3GDwZfPZjC/V0aeS1+Kb/NBeHaOTfEcqd3+nn4lYrEz3Aqi3NJoapgcWtcQDwIs9edRoJLxZ2iRJJqMY0ONTDur5qgAw+21Pk1KCHFMYVwHcXuOqf5fGTxhi0m1qSEHXJUNnVHP9mAwcF9YvmePeBi/s9RU/E6vHjjg7ZH//JTnGYERn9EqygXwFP+BKx+HrI+8mzmgOiZvwJkUrA6Yk6Z0QIcMiBfIVKm8AdQr4lo8UtZcm1YWz4z0wnN24UxKf3PwKy9Br9Ol2T4nLlAfRYBmNF6HM/nsC+TENfelZvp8xkH79hsdSfCNoWS5Ir1LgvKGJ/Nr/msM80bvBOqsqeT3MaInuWVRfAVhYlFhqUHuVSQf74qbSYXCyMnUR5GYZ6ZY/aYcXw1bFDmBLBRroMw4E76NLsmqoAU8Cq9PaZIqsGGt5fqv5tJMEPaAWr0tNwywtVVoLV3wZh15dClMRkfvrYEUuSBLw0neAHDVNz4UYvySfQgAtDIndxR9lwoBssMGi4zNVfFJ/kAFSTaWoiwu2Q7dp4LJIgG1ZHvUV1uICZhd9rPjEjS83FNCSAt1WBX7r8hLbBLhZBjl7dMl4CQqW4xC6JjzHjACiHGpjoyBBy9arrxg4PU= - + - SERVICE_NAME=async-http-client + - REVISION=$(git rev-parse --short ${TRAVIS_COMMIT}) + - secure: R2rTEfMHG3CHfbFBiSin6QI7hOm97DAIRz5bX0/My/jSuxGxblTbw3XntqFhByjbCgoWQn9crUQ1bdTp8RRmNC4/bbsdaPsXhD9jKnL/aJ+bNkEuh2/akyGnKvH300c2ZyGm5q5eC5FA20ErX5c3+86Sy821ywa/iC11YaV5XzP6lyxuFTJWs3e7SK7kbL4seyNd2sD/8DYzEeCwKxpCI8bVz7ZQUazP2IpkIjcS9SBeLwPHKbS1f2je8b6XIOkQUvJzxs3jrWIH9HwakA8/M/CSFo+GUkkRtPDj3V+lg1IyOA3VPLMxAVqeoryLIS6ITR8Zti+MDrodLM0llwo2dQ3ZI1RYzYieGcAg0XCflcH0fmrJd1gPiCuYKSmL48mAvgUkCThNU8nFekaqLLsspv7pS7+Ozi7f/j1pq8j4PkV2TV8HUa9RBcOepdeQnKxuFxc55NXfGvDPvqf9jw6CGxWqtFwRQg42gZtVWILgTvhmPchaIj0TKGYnnxG1jXXK0CQcMGOhF/T7WMrfZ1ZPVkT3/GUvh+PC7m2DE+dXonM1Ebo0p3lgWYlrlGv7fgjcrdHYi7FvqG0lZtUuNphGeoU7+QYOwZ0o7p7tboK24YW0dv7jybiCIzOloAfGgX+YflRZY9WpYJWMa0C9LaXok64oRT4/Fu0TEABHltc+TaY= + - secure: YK54XEeT5p/vCRG56GigVYAPVOEbqvHvltitS8JuhKw0jomYeB6FZf8N6ELZcbJh3pclwa9irihCGUxEbUKYaepQLBuns+c+EFKeEApFRdCUaZ/psKtg1A5gx+5MB4Qz5/ZA/CdyDYe3Muf0OAgkzfZxG9ayPtdanaOVCHyrY8jJDRCkgWqqOhnD1pw9MxhQpBUdtbOhdzb+F/mZ0P3be6+49BuFHcaPR3Ppw+3sUSy7ABoAWebjSk//FruEAPUel2WbivHo4GzJZydy4RqQG0YYkif2cz2h9I84uxLwXyl27dKd7oUvUtLaE1oQpYshoDGlCiO4VWUNDdMrrf3eBzkJ0QqF/wNF9CyYN1WIigK+gJ8vusPqB3V3f90RMdw6Q8vSLin0hcVpCTmDNXKlYmRJ5qzqWoyn7kX0vW4qCwCxiaL+iJg+Lu1ftIoWOY5oFmR6hkQjsIxrUqrTa2DA/SFU2Sde9E3Kx0q7l7BGJEcscmmyjrCJfFFaZX3jjY2So+xjgr0ymVO5KTG6PV8PZavsOG9TEXs8dl/9Bw8aup4rna8zixqp0nPJbQR9fLS9oub4eUpOIjl8fM3mbC7gAxr6zvQJ8yUdsBpjGiRHIWRzY2bAIDDSyqVo6uwlkeIlbXAakEgBFt6gcALvlOPV3YW6+rHps0hq6s4YYYXdhrE= + - secure: Ay+VkQA2gipR88zavvQ28IdrUFv5MQ9HSnIUxbt3iqjc1Qx2rmpRvC44L07LoU3gYzxeu5iyzp5qVjKgcuGeqTKZCJDCUI/iX0LyJZoZoZSnMRdNLblVie/Wt7kTn/jax5vdDR40+DJLDYBrTl9t+/Maut++gEJfeqbjYcIrS1GRC4XtQfgADu6yV5X4FwjaGRu6Ws6U3OAl4Obh9IsfGiB7ildpVGryXj2B348dneANCDT1JUpeWNJtGszr3HKSqgJFhAmDWYSoHluPZhMd1XofS18rNM3/YCQ+p078o4GGml9QPYAt5voMiJXYvOZl1FEehPPdDbEV0rpc12BLK8cq2a4uxWLYfE7SmQODHU2MtsBvqCNp7jEk4K00ub+B27Y4i45MSP6qWf4pZaACjg2RIDbGMIrSfZpxRodYFO6QgwobxS6kPrJQzBJJPjk82MIkvzxG1sw4BxAJBy+IctUlwLdeopXM+rUpjvfGy4Bv9QNHbma+3DA8SIL/hvHsdnKiFxVGgh7tdoWDSpKqENZptWUcPMx5OsnDT8cv4TcMiEVX5+Ya+rQYUB9qVfG34QOSnTCaCGrUIarfMSWNnFhNWDv6AqTPF3rlLxASPJ17OOnD0VMm+IfGLE6Kv7MBi99dXAqHHu1OgLbJhzCA3Wqtflkv8hZzn9CpANJbMgI= + - secure: eFF6Eyz3nluIT0nW0Dx0cCY5ho88CNZyxv8hXUUSTzuyDSfAIaPPIyTjgCNcoMxLWjEyR1NuNywcWUv50tJBeTv8U5jJSkP4SmZtMIqoFGEobOUuQ06eHPtdMhxZSjmgWbnET45AAG+lUBIwfWndItukaxGMOtuS6Lv4pjCNZeLXhLEJPeJ6OEsE1ZerqZojAFlv0Jge7D8RyVdx1pfHec85vBbpS9kfM/McntHRVDJuVsKjvIE1ExzL3L/wK4ouJxdC20rYdZv6q6zTEjbAVZWn3kDJy7jL+warL1eTv/YZda1+1TVSKFwZNgjmtV2CLIdpSlJIGmP4N3Svp/2wF+vMCEhOHB/KgiaPjefcjPT0CUOAEExoXaF2mccZysPazk8qvoQ3DU7o6BuXu/sIC5o/5qYzhlm/Ir/3xgWHsM5m7hTyol+emtt1Nhh+EGpAORIAfb6+g2acqYfgp4HsR14VPblYwd33PG3hF3aGqa5eKhX9lwYNZr98S6VyVu/1vRNyBpes8lcaVCKzQd6s5urmHQQge8HKESuJdNuLBn2m4F8/6TWgTwxywiMKezOwRVnKciJffMqkoXXmwuPEt+aHJyQp6e7zlo0pFxohNCiyxr/9Us6tXrXX6sIFVOc9uwBDYr53pRA5n3tUrJDeuLWCIGSQdB7G2ZUkI3+jtOI= cache: directories: - - build-cache - + - build-cache install: - - docker-compose -f docker-compose.yml build - +- docker-compose -f docker-compose.yml build script: - - docker-compose -f docker-compose.yml run ${SERVICE_NAME} - +- docker-compose -f docker-compose.yml run ${SERVICE_NAME} before_cache: - - docker-compose -f docker-compose.yml run --rm -v `pwd`/build-cache:/app/build-cache ${SERVICE_NAME} bash docker-utils.sh save-cache +- docker-compose -f docker-compose.yml run --rm -v `pwd`/build-cache:/app/build-cache + ${SERVICE_NAME} bash docker-utils.sh save-cache diff --git a/README.md b/README.md index 487cf8ffc5..6685707ca7 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ Follow [@AsyncHttpClient](https://twitter.com/AsyncHttpClient) on Twitter. The AsyncHttpClient (AHC) library allows Java applications to easily execute HTTP requests and asynchronously process HTTP responses. The library also supports the WebSocket Protocol. -It's built on top of [Netty](https://github.com/netty/netty). I's currently compiled on Java 8 but runs on Java 9 too. +It's built on top of [Netty](https://github.com/netty/netty). It's currently compiled on Java 8 but runs on Java 9 too. ## Installation @@ -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 @@ -158,7 +159,7 @@ See `AsyncCompletionHandler` implementation as an example. The below sample just capture the response status and skips processing the response body chunks. -Note that returning `ABORT` closed the underlying connection. +Note that returning `ABORT` closes the underlying connection. ```java import static org.asynchttpclient.Dsl.*; @@ -195,7 +196,7 @@ Integer statusCode = whenStatusCode.get(); #### Using Continuations -`ListenableFuture` has a `toCompletableFuture` that returns a `CompletableFuture`. +`ListenableFuture` has a `toCompletableFuture` method that returns a `CompletableFuture`. Beware that canceling this `CompletableFuture` won't properly cancel the ongoing request. There's a very good chance we'll return a `CompletionStage` instead in the next release. @@ -243,7 +244,7 @@ WebSocket websocket = c.prepareGet("ws://demos.kaazing.com/echo") ## Reactive Streams -AsyncHttpClient has build in support for reactive streams. +AsyncHttpClient has built-in support for reactive streams. You can pass a request body as a `Publisher` or a `ReactiveStreamsBodyGenerator`. @@ -288,7 +289,7 @@ Keep up to date on the library development by joining the Asynchronous HTTP Clie Of course, Pull Requests are welcome. -Here a the few rules we'd like you to respect if you do so: +Here are the few rules we'd like you to respect if you do so: * Only edit the code related to the suggested change, so DON'T automatically format the classes you've edited. * Use IntelliJ default formatting rules. diff --git a/client/pom.xml b/client/pom.xml index 5b635deae0..72a0e1fbf3 100644 --- a/client/pom.xml +++ b/client/pom.xml @@ -73,5 +73,10 @@ reactive-streams-examples test + + org.apache.kerby + kerb-simplekdc + test + diff --git a/client/src/main/java/org/asynchttpclient/AsyncCompletionHandler.java b/client/src/main/java/org/asynchttpclient/AsyncCompletionHandler.java index c7b60e0b4c..d1f30c1ac3 100644 --- a/client/src/main/java/org/asynchttpclient/AsyncCompletionHandler.java +++ b/client/src/main/java/org/asynchttpclient/AsyncCompletionHandler.java @@ -24,7 +24,7 @@ /** * An {@link AsyncHandler} augmented with an {@link #onCompleted(Response)} * convenience method which gets called when the {@link Response} processing is - * finished. This class also implement the {@link ProgressAsyncHandler} + * finished. This class also implements the {@link ProgressAsyncHandler} * callback, all doing nothing except returning * {@link org.asynchttpclient.AsyncHandler.State#CONTINUE} * diff --git a/client/src/main/java/org/asynchttpclient/AsyncHandler.java b/client/src/main/java/org/asynchttpclient/AsyncHandler.java index 090503cf14..6733c94711 100644 --- a/client/src/main/java/org/asynchttpclient/AsyncHandler.java +++ b/client/src/main/java/org/asynchttpclient/AsyncHandler.java @@ -19,6 +19,7 @@ import io.netty.handler.codec.http.HttpHeaders; import org.asynchttpclient.netty.request.NettyRequest; +import javax.net.ssl.SSLSession; import java.net.InetSocketAddress; import java.util.List; @@ -37,9 +38,9 @@ * *
* Returning a {@link AsyncHandler.State#ABORT} from any of those callback methods will interrupt asynchronous response - * processing, after that only {@link #onCompleted()} is going to be called. + * processing. After that, only {@link #onCompleted()} is going to be called. *
- * AsyncHandler aren't thread safe, hence you should avoid re-using the same instance when doing concurrent requests. + * AsyncHandlers aren't thread safe. Hence, you should avoid re-using the same instance when doing concurrent requests. * As an example, the following may produce unexpected results: * 
  *   AsyncHandler ah = new AsyncHandler() {....};
@@ -49,9 +50,10 @@
  *
* It is recommended to create a new instance instead. *

- * Do NOT perform any blocking operation in there, typically trying to send another request and call get() on its future. + * Do NOT perform any blocking operations in any of these methods. A typical example would be trying to send another + * request and calling get() on its future. * There's a chance you might end up in a dead lock. - * If you really to perform blocking operation, executed it in a different dedicated thread pool. + * If you really need to perform a blocking operation, execute it in a different dedicated thread pool. * * @param Type of object returned by the {@link java.util.concurrent.Future#get} */ @@ -142,6 +144,8 @@ default void onHostnameResolutionSuccess(String name, List ad default void onHostnameResolutionFailure(String name, Throwable cause) { } + // ////////////// TCP CONNECT //////// + /** * Notify the callback when trying to open a new connection. *

@@ -152,8 +156,6 @@ default void onHostnameResolutionFailure(String name, Throwable cause) { default void onTcpConnectAttempt(InetSocketAddress remoteAddress) { } - // ////////////// TCP CONNECT //////// - /** * Notify the callback after a successful connect * @@ -174,18 +176,18 @@ default void onTcpConnectSuccess(InetSocketAddress remoteAddress, Channel connec default void onTcpConnectFailure(InetSocketAddress remoteAddress, Throwable cause) { } + // ////////////// TLS /////////////// + /** * Notify the callback before TLS handshake */ default void onTlsHandshakeAttempt() { } - // ////////////// TLS /////////////// - /** * Notify the callback after the TLS was successful */ - default void onTlsHandshakeSuccess() { + default void onTlsHandshakeSuccess(SSLSession sslSession) { } /** @@ -196,14 +198,14 @@ default void onTlsHandshakeSuccess() { default void onTlsHandshakeFailure(Throwable cause) { } + // /////////// POOLING ///////////// + /** * Notify the callback when trying to fetch a connection from the pool. */ default void onConnectionPoolAttempt() { } - // /////////// POOLING ///////////// - /** * Notify the callback when a new connection was successfully fetched from the pool. * @@ -220,6 +222,8 @@ default void onConnectionPooled(Channel connection) { default void onConnectionOffer(Channel connection) { } + // //////////// SENDING ////////////// + /** * Notify the callback when a request is being written on the channel. If the original request causes multiple requests to be sent, for example, because of authorization or * retry, it will be notified multiple times. @@ -229,8 +233,6 @@ default void onConnectionOffer(Channel connection) { default void onRequestSend(NettyRequest request) { } - // //////////// SENDING ////////////// - /** * Notify the callback every time a request is being retried. */ diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClient.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClient.java index 1510de4513..2ab335f3f6 100755 --- a/client/src/main/java/org/asynchttpclient/AsyncHttpClient.java +++ b/client/src/main/java/org/asynchttpclient/AsyncHttpClient.java @@ -21,15 +21,15 @@ import java.util.function.Predicate; /** - * This class support asynchronous and synchronous HTTP request. + * This class support asynchronous and synchronous HTTP requests. *
- * To execute synchronous HTTP request, you just need to do + * To execute a synchronous HTTP request, you just need to do * 

  *    AsyncHttpClient c = new AsyncHttpClient();
  *    Future<Response> f = c.prepareGet(TARGET_URL).execute();
  *
*
- * The code above will block until the response is fully received. To execute asynchronous HTTP request, you + * The code above will block until the response is fully received. To execute an asynchronous HTTP request, you * create an {@link AsyncHandler} or its abstract implementation, {@link AsyncCompletionHandler} *
* 
@@ -48,7 +48,7 @@
  *      });
  *      Response response = f.get();
  *
- *      // We are just interested to retrieve the status code.
+ *      // We are just interested in retrieving the status code.
  *     Future<Integer> f = c.prepareGet(TARGET_URL).execute(new AsyncCompletionHandler<Integer>() {
  *
  *          @Override
@@ -63,10 +63,10 @@
  *      });
  *      Integer statusCode = f.get();
  *
- * The {@link AsyncCompletionHandler#onCompleted(Response)} will be invoked once the http response has been fully read, which include - * the http headers and the response body. Note that the entire response will be buffered in memory. + * The {@link AsyncCompletionHandler#onCompleted(Response)} method will be invoked once the http response has been fully read. + * The {@link Response} object includes the http headers and the response body. Note that the entire response will be buffered in memory. *
- * You can also have more control about the how the response is asynchronously processed by using a {@link AsyncHandler} + * You can also have more control about the how the response is asynchronously processed by using an {@link AsyncHandler} * 
  *      AsyncHttpClient c = new AsyncHttpClient();
  *      Future<String> f = c.prepareGet(TARGET_URL).execute(new AsyncHandler<String>() {
@@ -106,8 +106,8 @@
  *
  *      String bodyResponse = f.get();
  *
- * You can asynchronously process the response status,headers and body and decide when to - * stop the processing the response by returning a new {@link AsyncHandler.State#ABORT} at any moment. + * You can asynchronously process the response status, headers and body and decide when to + * stop processing the response by returning a new {@link AsyncHandler.State#ABORT} at any moment. * * This class can also be used without the need of {@link AsyncHandler}. *
@@ -125,8 +125,8 @@ * Response r = f.get(); * *
- * An instance of this class will cache every HTTP 1.1 connections and close them when the {@link DefaultAsyncHttpClientConfig#getReadTimeout()} - * expires. This object can hold many persistent connections to different host. + * An instance of this class will cache every HTTP 1.1 connection and close them when the {@link DefaultAsyncHttpClientConfig#getReadTimeout()} + * expires. This object can hold many persistent connections to different hosts. */ public interface AsyncHttpClient extends Closeable { @@ -138,7 +138,7 @@ public interface AsyncHttpClient extends Closeable { boolean isClosed(); /** - * Set default signature calculator to use for requests build by this client instance + * Set default signature calculator to use for requests built by this client instance * * @param signatureCalculator a signature calculator * @return {@link RequestBuilder} diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java index e0f8413662..862aa2ce9f 100644 --- a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java @@ -65,6 +65,14 @@ public interface AsyncHttpClientConfig { */ int getMaxConnectionsPerHost(); + /** + * Return the maximum duration in milliseconds an {@link AsyncHttpClient} can wait to acquire a free channel + * + * @return Return the maximum duration in milliseconds an {@link AsyncHttpClient} can wait to acquire a free channel + */ + int getAcquireFreeChannelTimeout(); + + /** * Return the maximum time in millisecond an {@link AsyncHttpClient} can wait when connecting to a remote host * diff --git a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java index 1827728e4c..d26612fb6d 100644 --- a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java @@ -84,6 +84,7 @@ public class DefaultAsyncHttpClientConfig implements AsyncHttpClientConfig { private final int connectionTtl; private final int maxConnections; private final int maxConnectionsPerHost; + private final int acquireFreeChannelTimeout; private final ChannelPool channelPool; private final ConnectionSemaphoreFactory connectionSemaphoreFactory; private final KeepAliveStrategy keepAliveStrategy; @@ -163,6 +164,7 @@ private DefaultAsyncHttpClientConfig(// http int connectionTtl, int maxConnections, int maxConnectionsPerHost, + int acquireFreeChannelTimeout, ChannelPool channelPool, ConnectionSemaphoreFactory connectionSemaphoreFactory, KeepAliveStrategy keepAliveStrategy, @@ -250,6 +252,7 @@ private DefaultAsyncHttpClientConfig(// http this.connectionTtl = connectionTtl; this.maxConnections = maxConnections; this.maxConnectionsPerHost = maxConnectionsPerHost; + this.acquireFreeChannelTimeout = acquireFreeChannelTimeout; this.channelPool = channelPool; this.connectionSemaphoreFactory = connectionSemaphoreFactory; this.keepAliveStrategy = keepAliveStrategy; @@ -445,6 +448,9 @@ public int getMaxConnectionsPerHost() { return maxConnectionsPerHost; } + @Override + public int getAcquireFreeChannelTimeout() { return acquireFreeChannelTimeout; } + @Override public ChannelPool getChannelPool() { return channelPool; @@ -696,6 +702,7 @@ public static class Builder { private int connectionTtl = defaultConnectionTtl(); private int maxConnections = defaultMaxConnections(); private int maxConnectionsPerHost = defaultMaxConnectionsPerHost(); + private int acquireFreeChannelTimeout = defaultAcquireFreeChannelTimeout(); private ChannelPool channelPool; private ConnectionSemaphoreFactory connectionSemaphoreFactory; private KeepAliveStrategy keepAliveStrategy = new DefaultKeepAliveStrategy(); @@ -754,6 +761,7 @@ public Builder(AsyncHttpClientConfig config) { realm = config.getRealm(); maxRequestRetry = config.getMaxRequestRetry(); disableUrlEncodingForBoundRequests = config.isDisableUrlEncodingForBoundRequests(); + useLaxCookieEncoder = config.isUseLaxCookieEncoder(); disableZeroCopy = config.isDisableZeroCopy(); keepEncodingHeader = config.isKeepEncodingHeader(); proxyServerSelector = config.getProxyServerSelector(); @@ -990,6 +998,16 @@ public Builder setMaxConnectionsPerHost(int maxConnectionsPerHost) { return this; } + /** + * Sets the maximum duration in milliseconds to acquire a free channel to send a request + * @param acquireFreeChannelTimeout maximum duration in milliseconds to acquire a free channel to send a request + * @return the same builder instance + */ + public Builder setAcquireFreeChannelTimeout(int acquireFreeChannelTimeout) { + this.acquireFreeChannelTimeout = acquireFreeChannelTimeout; + return this; + } + public Builder setChannelPool(ChannelPool channelPool) { this.channelPool = channelPool; return this; @@ -1248,6 +1266,7 @@ public DefaultAsyncHttpClientConfig build() { connectionTtl, maxConnections, maxConnectionsPerHost, + acquireFreeChannelTimeout, channelPool, connectionSemaphoreFactory, keepAliveStrategy, diff --git a/client/src/main/java/org/asynchttpclient/Dsl.java b/client/src/main/java/org/asynchttpclient/Dsl.java index 914b734b77..cdb30ed165 100644 --- a/client/src/main/java/org/asynchttpclient/Dsl.java +++ b/client/src/main/java/org/asynchttpclient/Dsl.java @@ -99,7 +99,11 @@ public static Realm.Builder realm(Realm prototype) { .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) { diff --git a/client/src/main/java/org/asynchttpclient/Realm.java b/client/src/main/java/org/asynchttpclient/Realm.java index 9b9bdf798e..c6324fd0b4 100644 --- a/client/src/main/java/org/asynchttpclient/Realm.java +++ b/client/src/main/java/org/asynchttpclient/Realm.java @@ -23,6 +23,7 @@ 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 customLoginConfig; + private final String servicePrincipalName; + private final boolean useCanonicalHostname; + private final String loginContextName; private Realm(AuthScheme scheme, String principal, @@ -78,11 +83,15 @@ private Realm(AuthScheme scheme, String ntlmDomain, String ntlmHost, boolean useAbsoluteURI, - boolean omitQuery) { + boolean omitQuery, + String servicePrincipalName, + boolean useCanonicalHostname, + Map 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 @@ private Realm(AuthScheme scheme, 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 boolean isOmitQuery() { return omitQuery; } + public Map 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 static class Builder { private String ntlmHost = "localhost"; private boolean useAbsoluteURI = false; private boolean omitQuery; + /** + * Kerberos/Spnego properties + */ + private Map 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 Builder setCharset(Charset charset) { return this; } + public Builder setCustomLoginConfig(Map 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 Realm build() { ntlmDomain, ntlmHost, useAbsoluteURI, - omitQuery); + omitQuery, + servicePrincipalName, + useCanonicalHostname, + customLoginConfig, + loginContextName); } } } diff --git a/client/src/main/java/org/asynchttpclient/RequestBuilder.java b/client/src/main/java/org/asynchttpclient/RequestBuilder.java index ad0a141495..4b0d485ba4 100644 --- a/client/src/main/java/org/asynchttpclient/RequestBuilder.java +++ b/client/src/main/java/org/asynchttpclient/RequestBuilder.java @@ -18,7 +18,7 @@ import static org.asynchttpclient.util.HttpConstants.Methods.GET; /** - * Builder for a {@link Request}. Warning: mutable and not thread-safe! Beware that it holds a reference on the Request instance it builds, so modifying the builder will modify the + * Builder for a {@link Request}. Warning: mutable and not thread-safe! Beware that it holds a reference to the Request instance it builds, so modifying the builder will modify the * request even after it has been built. */ public class RequestBuilder extends RequestBuilderBase { diff --git a/client/src/main/java/org/asynchttpclient/RequestBuilderBase.java b/client/src/main/java/org/asynchttpclient/RequestBuilderBase.java index 716b2ec7ac..79d2d990f7 100644 --- a/client/src/main/java/org/asynchttpclient/RequestBuilderBase.java +++ b/client/src/main/java/org/asynchttpclient/RequestBuilderBase.java @@ -270,7 +270,7 @@ public T setHeaders(HttpHeaders headers) { * @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> headers) { + public T setHeaders(Map> headers) { clearHeaders(); if (headers != null) { headers.forEach((name, values) -> this.headers.add(name, values)); @@ -285,7 +285,7 @@ public T setHeaders(Map> headers) { * @param headers map of header names as the map keys and header values as the map values * @return {@code this} */ - public T setSingleHeaders(Map headers) { + public T setSingleHeaders(Map headers) { clearHeaders(); if (headers != null) { headers.forEach((name, value) -> this.headers.add(name, value)); diff --git a/client/src/main/java/org/asynchttpclient/Response.java b/client/src/main/java/org/asynchttpclient/Response.java index f3bc6e3a91..99f033e995 100644 --- a/client/src/main/java/org/asynchttpclient/Response.java +++ b/client/src/main/java/org/asynchttpclient/Response.java @@ -147,24 +147,29 @@ public interface Response { boolean hasResponseHeaders(); /** - * Return true if the response's body has been computed by an {@link AsyncHandler}. It will return false if the either {@link AsyncHandler#onStatusReceived(HttpResponseStatus)} - * or {@link AsyncHandler#onHeadersReceived(HttpHeaders)} returned {@link AsyncHandler.State#ABORT} + * Return true if the response's body has been computed by an {@link AsyncHandler}. + * It will return false if: + *
    + *
  • either the {@link AsyncHandler#onStatusReceived(HttpResponseStatus)} returned {@link AsyncHandler.State#ABORT}
    • + *
  • or {@link AsyncHandler#onHeadersReceived(HttpHeaders)} returned {@link AsyncHandler.State#ABORT}
    • + *
  • response body was empty
    • + *
* - * @return true if the response's body has been computed by an {@link AsyncHandler} + * @return true if the response's body has been computed by an {@link AsyncHandler} to new empty bytes */ boolean hasResponseBody(); /** - * Get remote address client initiated request to. + * Get the remote address that the client initiated the request to. * - * @return remote address client initiated request to, may be {@code null} if asynchronous provider is unable to provide the remote address + * @return The remote address that the client initiated the request to. May be {@code null} if asynchronous provider is unable to provide the remote address */ SocketAddress getRemoteAddress(); /** - * Get local address client initiated request from. + * Get the local address that the client initiated the request from. * - * @return local address client initiated request from, may be {@code null} if asynchronous provider is unable to provide the local address + * @return The local address that the client initiated the request from. May be {@code null} if asynchronous provider is unable to provide the local address */ SocketAddress getLocalAddress(); diff --git a/client/src/main/java/org/asynchttpclient/SslEngineFactory.java b/client/src/main/java/org/asynchttpclient/SslEngineFactory.java index 1157e499f3..7fb25dd844 100644 --- a/client/src/main/java/org/asynchttpclient/SslEngineFactory.java +++ b/client/src/main/java/org/asynchttpclient/SslEngineFactory.java @@ -19,7 +19,7 @@ public interface SslEngineFactory { /** - * Creates new {@link SSLEngine}. + * Creates a new {@link SSLEngine}. * * @param config the client config * @param peerHost the peer hostname diff --git a/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java b/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java index 274537a6ad..fa073bc82f 100644 --- a/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java +++ b/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java @@ -22,6 +22,7 @@ public final class AsyncHttpClientConfigDefaults { public static final String THREAD_POOL_NAME_CONFIG = "threadPoolName"; public static final String MAX_CONNECTIONS_CONFIG = "maxConnections"; public static final String MAX_CONNECTIONS_PER_HOST_CONFIG = "maxConnectionsPerHost"; + public static final String ACQUIRE_FREE_CHANNEL_TIMEOUT = "acquireFreeChannelTimeout"; public static final String CONNECTION_TIMEOUT_CONFIG = "connectTimeout"; public static final String POOLED_CONNECTION_IDLE_TIMEOUT_CONFIG = "pooledConnectionIdleTimeout"; public static final String CONNECTION_POOL_CLEANER_PERIOD_CONFIG = "connectionPoolCleanerPeriod"; @@ -39,7 +40,7 @@ public final class AsyncHttpClientConfigDefaults { public static final String USE_PROXY_PROPERTIES_CONFIG = "useProxyProperties"; public static final String VALIDATE_RESPONSE_HEADERS_CONFIG = "validateResponseHeaders"; public static final String AGGREGATE_WEBSOCKET_FRAME_FRAGMENTS_CONFIG = "aggregateWebSocketFrameFragments"; - public static final String ENABLE_WEBSOCKET_COMPRESSION_CONFIG= "enableWebSocketCompression"; + public static final String ENABLE_WEBSOCKET_COMPRESSION_CONFIG = "enableWebSocketCompression"; public static final String STRICT_302_HANDLING_CONFIG = "strict302Handling"; public static final String KEEP_ALIVE_CONFIG = "keepAlive"; public static final String MAX_REQUEST_RETRY_CONFIG = "maxRequestRetry"; @@ -97,6 +98,10 @@ public static int defaultMaxConnectionsPerHost() { return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getInt(ASYNC_CLIENT_CONFIG_ROOT + MAX_CONNECTIONS_PER_HOST_CONFIG); } + public static int defaultAcquireFreeChannelTimeout() { + return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getInt(ASYNC_CLIENT_CONFIG_ROOT + ACQUIRE_FREE_CHANNEL_TIMEOUT); + } + public static int defaultConnectTimeout() { return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getInt(ASYNC_CLIENT_CONFIG_ROOT + CONNECTION_TIMEOUT_CONFIG); } diff --git a/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java b/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java index cb3b05fbd8..9f84ada7ab 100755 --- a/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java +++ b/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java @@ -90,7 +90,6 @@ public final class NettyResponseFuture implements ListenableFuture { private volatile int isCancelled = 0; private volatile int inAuth = 0; private volatile int inProxyAuth = 0; - private volatile int statusReceived = 0; @SuppressWarnings("unused") private volatile int contentProcessed = 0; @SuppressWarnings("unused") @@ -539,7 +538,6 @@ public String toString() { ",\n\tredirectCount=" + redirectCount + // ",\n\ttimeoutsHolder=" + TIMEOUTS_HOLDER_FIELD.get(this) + // ",\n\tinAuth=" + inAuth + // - ",\n\tstatusReceived=" + statusReceived + // ",\n\ttouch=" + touch + // '}'; } diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java index 8deb22e53a..6a5ed05972 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java @@ -345,7 +345,7 @@ public Future updatePipelineForHttpTunneling(ChannelPipeline pipeline, if (!isSslHandlerConfigured(pipeline)) { SslHandler sslHandler = createSslHandler(requestUri.getHost(), requestUri.getExplicitPort()); whenHanshaked = sslHandler.handshakeFuture(); - pipeline.addBefore(AHC_HTTP_HANDLER, SSL_HANDLER, sslHandler); + pipeline.addBefore(INFLATER_HANDLER, SSL_HANDLER, sslHandler); } pipeline.addAfter(SSL_HANDLER, HTTP_CLIENT_CODEC, newHttpClientCodec()); @@ -380,10 +380,11 @@ public SslHandler addSslHandler(ChannelPipeline pipeline, Uri uri, String virtua } SslHandler sslHandler = createSslHandler(peerHost, peerPort); - if (hasSocksProxyHandler) + if (hasSocksProxyHandler) { pipeline.addAfter(SOCKS_HANDLER, SSL_HANDLER, sslHandler); - else + } else { pipeline.addFirst(SSL_HANDLER, sslHandler); + } return sslHandler; } diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/CombinedConnectionSemaphore.java b/client/src/main/java/org/asynchttpclient/netty/channel/CombinedConnectionSemaphore.java new file mode 100644 index 0000000000..04549fd80d --- /dev/null +++ b/client/src/main/java/org/asynchttpclient/netty/channel/CombinedConnectionSemaphore.java @@ -0,0 +1,69 @@ +/* + * 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.netty.channel; + +import java.io.IOException; +import java.util.concurrent.TimeUnit; + +/** + * A combined {@link ConnectionSemaphore} with two limits - a global limit and a per-host limit + */ +public class CombinedConnectionSemaphore extends PerHostConnectionSemaphore { + protected final MaxConnectionSemaphore globalMaxConnectionSemaphore; + + CombinedConnectionSemaphore(int maxConnections, int maxConnectionsPerHost, int acquireTimeout) { + super(maxConnectionsPerHost, acquireTimeout); + this.globalMaxConnectionSemaphore = new MaxConnectionSemaphore(maxConnections, acquireTimeout); + } + + @Override + public void acquireChannelLock(Object partitionKey) throws IOException { + long remainingTime = super.acquireTimeout > 0 ? acquireGlobalTimed(partitionKey) : acquireGlobal(partitionKey); + + try { + if (remainingTime < 0 || !getFreeConnectionsForHost(partitionKey).tryAcquire(remainingTime, TimeUnit.MILLISECONDS)) { + releaseGlobal(partitionKey); + throw tooManyConnectionsPerHost; + } + } catch (InterruptedException e) { + releaseGlobal(partitionKey); + throw new RuntimeException(e); + } + } + + protected void releaseGlobal(Object partitionKey) { + this.globalMaxConnectionSemaphore.releaseChannelLock(partitionKey); + } + + protected long acquireGlobal(Object partitionKey) throws IOException { + this.globalMaxConnectionSemaphore.acquireChannelLock(partitionKey); + return 0; + } + + /* + * Acquires the global lock and returns the remaining time, in millis, to acquire the per-host lock + */ + protected long acquireGlobalTimed(Object partitionKey) throws IOException { + long beforeGlobalAcquire = System.currentTimeMillis(); + acquireGlobal(partitionKey); + long lockTime = System.currentTimeMillis() - beforeGlobalAcquire; + return this.acquireTimeout - lockTime; + } + + @Override + public void releaseChannelLock(Object partitionKey) { + this.globalMaxConnectionSemaphore.releaseChannelLock(partitionKey); + super.releaseChannelLock(partitionKey); + } +} diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/DefaultConnectionSemaphoreFactory.java b/client/src/main/java/org/asynchttpclient/netty/channel/DefaultConnectionSemaphoreFactory.java index a102f1def8..eba42186ee 100644 --- a/client/src/main/java/org/asynchttpclient/netty/channel/DefaultConnectionSemaphoreFactory.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/DefaultConnectionSemaphoreFactory.java @@ -17,14 +17,21 @@ public class DefaultConnectionSemaphoreFactory implements ConnectionSemaphoreFactory { - public ConnectionSemaphore newConnectionSemaphore(AsyncHttpClientConfig config) { - ConnectionSemaphore semaphore = new NoopConnectionSemaphore(); - if (config.getMaxConnections() > 0) { - semaphore = new MaxConnectionSemaphore(config.getMaxConnections()); - } - if (config.getMaxConnectionsPerHost() > 0) { - semaphore = new PerHostConnectionSemaphore(config.getMaxConnectionsPerHost(), semaphore); - } - return semaphore; + public ConnectionSemaphore newConnectionSemaphore(AsyncHttpClientConfig config) { + int acquireFreeChannelTimeout = Math.max(0, config.getAcquireFreeChannelTimeout()); + int maxConnections = config.getMaxConnections(); + int maxConnectionsPerHost = config.getMaxConnectionsPerHost(); + + if (maxConnections > 0 && maxConnectionsPerHost > 0) { + return new CombinedConnectionSemaphore(maxConnections, maxConnectionsPerHost, acquireFreeChannelTimeout); + } + if (maxConnections > 0) { + return new MaxConnectionSemaphore(maxConnections, acquireFreeChannelTimeout); } + if (maxConnectionsPerHost > 0) { + return new CombinedConnectionSemaphore(maxConnections, maxConnectionsPerHost, acquireFreeChannelTimeout); + } + + return new NoopConnectionSemaphore(); + } } diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/InfiniteSemaphore.java b/client/src/main/java/org/asynchttpclient/netty/channel/InfiniteSemaphore.java new file mode 100644 index 0000000000..97b8224739 --- /dev/null +++ b/client/src/main/java/org/asynchttpclient/netty/channel/InfiniteSemaphore.java @@ -0,0 +1,110 @@ +/* + * 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.netty.channel; + +import java.util.Collection; +import java.util.Collections; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; + +/** + * A java.util.concurrent.Semaphore that always has Integer.Integer.MAX_VALUE free permits + * + * @author Alex Maltinsky + */ +public class InfiniteSemaphore extends Semaphore { + + public static final InfiniteSemaphore INSTANCE = new InfiniteSemaphore(); + private static final long serialVersionUID = 1L; + + private InfiniteSemaphore() { + super(Integer.MAX_VALUE); + } + + @Override + public void acquire() { + // NO-OP + } + + @Override + public void acquireUninterruptibly() { + // NO-OP + } + + @Override + public boolean tryAcquire() { + return true; + } + + @Override + public boolean tryAcquire(long timeout, TimeUnit unit) { + return true; + } + + @Override + public void release() { + // NO-OP + } + + @Override + public void acquire(int permits) { + // NO-OP + } + + @Override + public void acquireUninterruptibly(int permits) { + // NO-OP + } + + @Override + public boolean tryAcquire(int permits) { + return true; + } + + @Override + public boolean tryAcquire(int permits, long timeout, TimeUnit unit) { + return true; + } + + @Override + public void release(int permits) { + // NO-OP + } + + @Override + public int availablePermits() { + return Integer.MAX_VALUE; + } + + @Override + public int drainPermits() { + return Integer.MAX_VALUE; + } + + @Override + protected void reducePermits(int reduction) { + // NO-OP + } + + @Override + public boolean isFair() { + return true; + } + + @Override + protected Collection getQueuedThreads() { + return Collections.emptyList(); + } +} + diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/MaxConnectionSemaphore.java b/client/src/main/java/org/asynchttpclient/netty/channel/MaxConnectionSemaphore.java index 99bd6a4be4..99c318afac 100644 --- a/client/src/main/java/org/asynchttpclient/netty/channel/MaxConnectionSemaphore.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/MaxConnectionSemaphore.java @@ -16,6 +16,8 @@ import org.asynchttpclient.exception.TooManyConnectionsException; import java.io.IOException; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; import static org.asynchttpclient.util.ThrowableUtil.unknownStackTrace; @@ -23,21 +25,29 @@ * Max connections limiter. * * @author Stepan Koltsov + * @author Alex Maltinsky */ public class MaxConnectionSemaphore implements ConnectionSemaphore { - private final NonBlockingSemaphoreLike freeChannels; - private final IOException tooManyConnections; + protected final Semaphore freeChannels; + protected final IOException tooManyConnections; + protected final int acquireTimeout; - MaxConnectionSemaphore(int maxConnections) { + MaxConnectionSemaphore(int maxConnections, int acquireTimeout) { tooManyConnections = unknownStackTrace(new TooManyConnectionsException(maxConnections), MaxConnectionSemaphore.class, "acquireChannelLock"); - freeChannels = maxConnections > 0 ? new NonBlockingSemaphore(maxConnections) : NonBlockingSemaphoreInfinite.INSTANCE; + freeChannels = maxConnections > 0 ? new Semaphore(maxConnections) : InfiniteSemaphore.INSTANCE; + this.acquireTimeout = Math.max(0, acquireTimeout); } @Override public void acquireChannelLock(Object partitionKey) throws IOException { - if (!freeChannels.tryAcquire()) - throw tooManyConnections; + try { + if (!freeChannels.tryAcquire(acquireTimeout, TimeUnit.MILLISECONDS)) { + throw tooManyConnections; + } + } catch (InterruptedException e) { + throw new RuntimeException(e); + } } @Override diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java b/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java index 76bd652a44..4a6f4dce20 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java @@ -130,7 +130,7 @@ public void onSuccess(Channel channel, InetSocketAddress remoteAddress) { @Override protected void onSuccess(Channel value) { try { - asyncHandler.onTlsHandshakeSuccess(); + asyncHandler.onTlsHandshakeSuccess(sslHandler.engine().getSession()); } catch (Exception e) { LOGGER.error("onTlsHandshakeSuccess crashed", e); NettyConnectListener.this.onFailure(channel, e); diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/NonBlockingSemaphore.java b/client/src/main/java/org/asynchttpclient/netty/channel/NonBlockingSemaphore.java deleted file mode 100644 index a7bd2eacfe..0000000000 --- a/client/src/main/java/org/asynchttpclient/netty/channel/NonBlockingSemaphore.java +++ /dev/null @@ -1,54 +0,0 @@ -/* - * Copyright (c) 2017 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.netty.channel; - -import java.util.concurrent.atomic.AtomicInteger; - -/** - * Semaphore-like API, but without blocking. - * - * @author Stepan Koltsov - */ -class NonBlockingSemaphore implements NonBlockingSemaphoreLike { - - private final AtomicInteger permits; - - NonBlockingSemaphore(int permits) { - this.permits = new AtomicInteger(permits); - } - - @Override - public void release() { - permits.incrementAndGet(); - } - - @Override - public boolean tryAcquire() { - for (; ; ) { - int count = permits.get(); - if (count <= 0) { - return false; - } - if (permits.compareAndSet(count, count - 1)) { - return true; - } - } - } - - @Override - public String toString() { - // mimic toString of Semaphore class - return super.toString() + "[Permits = " + permits + "]"; - } -} diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/NonBlockingSemaphoreInfinite.java b/client/src/main/java/org/asynchttpclient/netty/channel/NonBlockingSemaphoreInfinite.java deleted file mode 100644 index 3d4fb91dbd..0000000000 --- a/client/src/main/java/org/asynchttpclient/netty/channel/NonBlockingSemaphoreInfinite.java +++ /dev/null @@ -1,39 +0,0 @@ -/* - * Copyright (c) 2017 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.netty.channel; - -/** - * Non-blocking semaphore-like object with infinite permits. - *

- * So try-acquire always succeeds. - * - * @author Stepan Koltsov - */ -enum NonBlockingSemaphoreInfinite implements NonBlockingSemaphoreLike { - INSTANCE; - - @Override - public void release() { - } - - @Override - public boolean tryAcquire() { - return true; - } - - @Override - public String toString() { - return NonBlockingSemaphore.class.getName(); - } -} diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/NonBlockingSemaphoreLike.java b/client/src/main/java/org/asynchttpclient/netty/channel/NonBlockingSemaphoreLike.java deleted file mode 100644 index 44303c9dfc..0000000000 --- a/client/src/main/java/org/asynchttpclient/netty/channel/NonBlockingSemaphoreLike.java +++ /dev/null @@ -1,25 +0,0 @@ -/* - * Copyright (c) 2017 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.netty.channel; - -/** - * Non-blocking semaphore API. - * - * @author Stepan Koltsov - */ -interface NonBlockingSemaphoreLike { - void release(); - - boolean tryAcquire(); -} diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/PerHostConnectionSemaphore.java b/client/src/main/java/org/asynchttpclient/netty/channel/PerHostConnectionSemaphore.java index 5ebb348abf..9ce1f20e93 100644 --- a/client/src/main/java/org/asynchttpclient/netty/channel/PerHostConnectionSemaphore.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/PerHostConnectionSemaphore.java @@ -17,6 +17,8 @@ import java.io.IOException; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; import static org.asynchttpclient.util.ThrowableUtil.unknownStackTrace; @@ -25,37 +27,36 @@ */ public class PerHostConnectionSemaphore implements ConnectionSemaphore { - private final ConnectionSemaphore globalSemaphore; + protected final ConcurrentHashMap freeChannelsPerHost = new ConcurrentHashMap<>(); + protected final int maxConnectionsPerHost; + protected final IOException tooManyConnectionsPerHost; + protected final int acquireTimeout; - private final ConcurrentHashMap freeChannelsPerHost = new ConcurrentHashMap<>(); - private final int maxConnectionsPerHost; - private final IOException tooManyConnectionsPerHost; - - PerHostConnectionSemaphore(int maxConnectionsPerHost, ConnectionSemaphore globalSemaphore) { - this.globalSemaphore = globalSemaphore; + PerHostConnectionSemaphore(int maxConnectionsPerHost, int acquireTimeout) { tooManyConnectionsPerHost = unknownStackTrace(new TooManyConnectionsPerHostException(maxConnectionsPerHost), PerHostConnectionSemaphore.class, "acquireChannelLock"); this.maxConnectionsPerHost = maxConnectionsPerHost; + this.acquireTimeout = Math.max(0, acquireTimeout); } @Override public void acquireChannelLock(Object partitionKey) throws IOException { - globalSemaphore.acquireChannelLock(partitionKey); - - if (!getFreeConnectionsForHost(partitionKey).tryAcquire()) { - globalSemaphore.releaseChannelLock(partitionKey); - throw tooManyConnectionsPerHost; + try { + if (!getFreeConnectionsForHost(partitionKey).tryAcquire(acquireTimeout, TimeUnit.MILLISECONDS)) { + throw tooManyConnectionsPerHost; + } + } catch (InterruptedException e) { + throw new RuntimeException(e); } } @Override public void releaseChannelLock(Object partitionKey) { - globalSemaphore.releaseChannelLock(partitionKey); getFreeConnectionsForHost(partitionKey).release(); } - private NonBlockingSemaphoreLike getFreeConnectionsForHost(Object partitionKey) { + protected Semaphore getFreeConnectionsForHost(Object partitionKey) { return maxConnectionsPerHost > 0 ? - freeChannelsPerHost.computeIfAbsent(partitionKey, pk -> new NonBlockingSemaphore(maxConnectionsPerHost)) : - NonBlockingSemaphoreInfinite.INSTANCE; + freeChannelsPerHost.computeIfAbsent(partitionKey, pk -> new Semaphore(maxConnectionsPerHost)) : + InfiniteSemaphore.INSTANCE; } } diff --git a/client/src/main/java/org/asynchttpclient/netty/handler/AsyncHttpClientHandler.java b/client/src/main/java/org/asynchttpclient/netty/handler/AsyncHttpClientHandler.java index de78537ac7..ec158673f0 100755 --- a/client/src/main/java/org/asynchttpclient/netty/handler/AsyncHttpClientHandler.java +++ b/client/src/main/java/org/asynchttpclient/netty/handler/AsyncHttpClientHandler.java @@ -67,8 +67,10 @@ public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exce Object attribute = Channels.getAttribute(channel); try { - if (attribute instanceof OnLastHttpContentCallback && msg instanceof LastHttpContent) { - ((OnLastHttpContentCallback) attribute).call(); + if (attribute instanceof OnLastHttpContentCallback) { + if (msg instanceof LastHttpContent) { + ((OnLastHttpContentCallback) attribute).call(); + } } else if (attribute instanceof NettyResponseFuture) { NettyResponseFuture future = (NettyResponseFuture) attribute; diff --git a/client/src/main/java/org/asynchttpclient/netty/handler/intercept/ProxyUnauthorized407Interceptor.java b/client/src/main/java/org/asynchttpclient/netty/handler/intercept/ProxyUnauthorized407Interceptor.java index 0812083ad5..02ee195622 100644 --- a/client/src/main/java/org/asynchttpclient/netty/handler/intercept/ProxyUnauthorized407Interceptor.java +++ b/client/src/main/java/org/asynchttpclient/netty/handler/intercept/ProxyUnauthorized407Interceptor.java @@ -140,7 +140,7 @@ public boolean exitAfterHandling407(Channel channel, return false; } try { - kerberosProxyChallenge(proxyServer, requestHeaders); + kerberosProxyChallenge(proxyRealm, proxyServer, requestHeaders); } catch (SpnegoEngineException e) { // FIXME @@ -184,10 +184,17 @@ public boolean exitAfterHandling407(Channel channel, 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); } diff --git a/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Unauthorized401Interceptor.java b/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Unauthorized401Interceptor.java index e63daece58..30ba1bc3d6 100644 --- a/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Unauthorized401Interceptor.java +++ b/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Unauthorized401Interceptor.java @@ -139,7 +139,7 @@ public boolean exitAfterHandling401(final Channel channel, return false; } try { - kerberosChallenge(request, requestHeaders); + kerberosChallenge(realm, request, requestHeaders); } catch (SpnegoEngineException e) { // FIXME @@ -200,12 +200,19 @@ private void ntlmChallenge(String authenticateHeader, } } - 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); } } diff --git a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyBodyBody.java b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyBodyBody.java index 728e2ec896..1a7d50b3fd 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyBodyBody.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyBodyBody.java @@ -53,7 +53,7 @@ public long getContentLength() { 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 { diff --git a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyDirectBody.java b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyDirectBody.java index 0d25358713..9d4eacb165 100644 --- a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyDirectBody.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyDirectBody.java @@ -23,6 +23,6 @@ public abstract class NettyDirectBody implements NettyBody { @Override public void write(Channel channel, NettyResponseFuture future) { - throw new UnsupportedOperationException("This kind of body is supposed to be writen directly"); + throw new UnsupportedOperationException("This kind of body is supposed to be written directly"); } } diff --git a/client/src/main/java/org/asynchttpclient/oauth/OAuthSignatureCalculatorInstance.java b/client/src/main/java/org/asynchttpclient/oauth/OAuthSignatureCalculatorInstance.java index acb9fce5b1..aa92c5aaa1 100644 --- a/client/src/main/java/org/asynchttpclient/oauth/OAuthSignatureCalculatorInstance.java +++ b/client/src/main/java/org/asynchttpclient/oauth/OAuthSignatureCalculatorInstance.java @@ -199,7 +199,6 @@ private byte[] digest(ConsumerKey consumerAuth, RequestToken userAuth, ByteBuffe SecretKeySpec signingKey = new SecretKeySpec(keyBytes, HMAC_SHA1_ALGORITHM); mac.init(signingKey); - mac.reset(); mac.update(message); return mac.doFinal(); } diff --git a/client/src/main/java/org/asynchttpclient/request/body/multipart/InputStreamPart.java b/client/src/main/java/org/asynchttpclient/request/body/multipart/InputStreamPart.java new file mode 100644 index 0000000000..ca7d0db367 --- /dev/null +++ b/client/src/main/java/org/asynchttpclient/request/body/multipart/InputStreamPart.java @@ -0,0 +1,66 @@ +/* + * 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; + } +} diff --git a/client/src/main/java/org/asynchttpclient/request/body/multipart/MultipartUtils.java b/client/src/main/java/org/asynchttpclient/request/body/multipart/MultipartUtils.java index 94bcb295d5..78e2d130a4 100644 --- a/client/src/main/java/org/asynchttpclient/request/body/multipart/MultipartUtils.java +++ b/client/src/main/java/org/asynchttpclient/request/body/multipart/MultipartUtils.java @@ -75,6 +75,9 @@ public static List> generateMultipartParts(List { + + 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); + } + +} diff --git a/client/src/main/java/org/asynchttpclient/request/body/multipart/part/MultipartPart.java b/client/src/main/java/org/asynchttpclient/request/body/multipart/part/MultipartPart.java index 38041338e8..b8c8622680 100644 --- a/client/src/main/java/org/asynchttpclient/request/body/multipart/part/MultipartPart.java +++ b/client/src/main/java/org/asynchttpclient/request/body/multipart/part/MultipartPart.java @@ -106,6 +106,10 @@ public abstract class MultipartPart implements Closeable { } public long length() { + long contentLength = getContentLength(); + if (contentLength < 0) { + return contentLength; + } return preContentLength + postContentLength + getContentLength(); } diff --git a/client/src/main/java/org/asynchttpclient/spnego/NamePasswordCallbackHandler.java b/client/src/main/java/org/asynchttpclient/spnego/NamePasswordCallbackHandler.java new file mode 100644 index 0000000000..ba79f9883a --- /dev/null +++ b/client/src/main/java/org/asynchttpclient/spnego/NamePasswordCallbackHandler.java @@ -0,0 +1,82 @@ +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 diff --git a/client/src/main/java/org/asynchttpclient/spnego/SpnegoEngine.java b/client/src/main/java/org/asynchttpclient/spnego/SpnegoEngine.java index 3326dca931..515bf63184 100644 --- a/client/src/main/java/org/asynchttpclient/spnego/SpnegoEngine.java +++ b/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.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 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 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 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 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,11 +167,30 @@ public String generateToken(String server) throws SpnegoEngineException { 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, - GSSContext.DEFAULT_LIFETIME); + 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 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); } catch (GSSException ex) { @@ -123,7 +210,7 @@ public String generateToken(String server) throws SpnegoEngineException { 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 String generateToken(String server) throws SpnegoEngineException { 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; + } } diff --git a/client/src/main/java/org/asynchttpclient/util/AuthenticatorUtils.java b/client/src/main/java/org/asynchttpclient/util/AuthenticatorUtils.java index 59754e22a8..00d69af7d2 100644 --- a/client/src/main/java/org/asynchttpclient/util/AuthenticatorUtils.java +++ b/client/src/main/java/org/asynchttpclient/util/AuthenticatorUtils.java @@ -175,7 +175,14 @@ else if (request.getVirtualHost() != null) 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); } diff --git a/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties b/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties index cdc632f701..c6fb355d75 100644 --- a/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties +++ b/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties @@ -1,6 +1,7 @@ org.asynchttpclient.threadPoolName=AsyncHttpClient org.asynchttpclient.maxConnections=-1 org.asynchttpclient.maxConnectionsPerHost=-1 +org.asynchttpclient.acquireFreeChannelTimeout=0 org.asynchttpclient.connectTimeout=5000 org.asynchttpclient.pooledConnectionIdleTimeout=60000 org.asynchttpclient.connectionPoolCleanerPeriod=1000 diff --git a/client/src/test/java/org/asynchttpclient/AsyncStreamHandlerTest.java b/client/src/test/java/org/asynchttpclient/AsyncStreamHandlerTest.java index e5ec906576..1547872aaa 100644 --- a/client/src/test/java/org/asynchttpclient/AsyncStreamHandlerTest.java +++ b/client/src/test/java/org/asynchttpclient/AsyncStreamHandlerTest.java @@ -25,6 +25,8 @@ import org.testng.annotations.Test; import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @@ -40,7 +42,7 @@ public class AsyncStreamHandlerTest extends HttpTest { - private static final String RESPONSE = "param_1_"; + private static final String RESPONSE = "param_1=value_1"; private static HttpServer server; @@ -93,18 +95,25 @@ public void asyncStreamPOSTTest() throws Throwable { @Override public State onHeadersReceived(HttpHeaders headers) { assertContentTypesEquals(headers.get(CONTENT_TYPE), TEXT_HTML_CONTENT_TYPE_WITH_UTF_8_CHARSET); + for (Map.Entry header : headers) { + if (header.getKey().startsWith("X-param")) { + builder.append(header.getKey().substring(2)).append("=").append(header.getValue()).append("&"); + } + } return State.CONTINUE; } @Override public State onBodyPartReceived(HttpResponseBodyPart content) { - builder.append(new String(content.getBodyPartBytes(), US_ASCII)); return State.CONTINUE; } @Override public String onCompleted() { - return builder.toString().trim(); + if (builder.length() > 0) { + builder.setLength(builder.length() - 1); + } + return builder.toString(); } }).get(10, TimeUnit.SECONDS); @@ -174,17 +183,24 @@ public void asyncStreamFutureTest() throws Throwable { public State onHeadersReceived(HttpHeaders headers) { assertContentTypesEquals(headers.get(CONTENT_TYPE), TEXT_HTML_CONTENT_TYPE_WITH_UTF_8_CHARSET); onHeadersReceived.set(true); + for (Map.Entry header : headers) { + if (header.getKey().startsWith("X-param")) { + builder.append(header.getKey().substring(2)).append("=").append(header.getValue()).append("&"); + } + } return State.CONTINUE; } @Override public State onBodyPartReceived(HttpResponseBodyPart content) { - builder.append(new String(content.getBodyPartBytes())); return State.CONTINUE; } @Override public String onCompleted() { + if (builder.length() > 0) { + builder.setLength(builder.length() - 1); + } return builder.toString().trim(); } @@ -254,17 +270,24 @@ public void asyncStreamReusePOSTTest() throws Throwable { @Override public State onHeadersReceived(HttpHeaders headers) { responseHeaders.set(headers); + for (Map.Entry header : headers) { + if (header.getKey().startsWith("X-param")) { + builder.append(header.getKey().substring(2)).append("=").append(header.getValue()).append("&"); + } + } return State.CONTINUE; } @Override public State onBodyPartReceived(HttpResponseBodyPart content) { - builder.append(new String(content.getBodyPartBytes())); return State.CONTINUE; } @Override public String onCompleted() { + if (builder.length() > 0) { + builder.setLength(builder.length() - 1); + } return builder.toString(); } }); diff --git a/client/src/test/java/org/asynchttpclient/BasicHttpTest.java b/client/src/test/java/org/asynchttpclient/BasicHttpTest.java index 0a26310491..d38c930f91 100755 --- a/client/src/test/java/org/asynchttpclient/BasicHttpTest.java +++ b/client/src/test/java/org/asynchttpclient/BasicHttpTest.java @@ -38,7 +38,10 @@ import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; +import java.io.UnsupportedEncodingException; import java.net.ConnectException; +import java.net.URLDecoder; +import java.net.URLEncoder; import java.net.UnknownHostException; import java.nio.charset.StandardCharsets; import java.util.*; @@ -206,6 +209,40 @@ public Response onCompleted(Response response) { })); } + @Test + public void postChineseChar() throws Throwable { + withClient().run(client -> + withServer(server).run(server -> { + HttpHeaders h = new DefaultHttpHeaders(); + h.add(CONTENT_TYPE, HttpHeaderValues.APPLICATION_X_WWW_FORM_URLENCODED); + + String chineseChar = "是"; + + Map> m = new HashMap<>(); + m.put("param", Collections.singletonList(chineseChar)); + + Request request = post(getTargetUrl()).setHeaders(h).setFormParams(m).build(); + + server.enqueueEcho(); + + client.executeRequest(request, new AsyncCompletionHandlerAdapter() { + @Override + public Response onCompleted(Response response) { + assertEquals(response.getStatusCode(), 200); + String value; + try { + // headers must be encoded + value = URLDecoder.decode(response.getHeader("X-param"), StandardCharsets.UTF_8.name()); + } catch (UnsupportedEncodingException e) { + throw new RuntimeException(e); + } + assertEquals(value, chineseChar); + return response; + } + }).get(TIMEOUT, SECONDS); + })); + } + @Test public void headHasEmptyBody() throws Throwable { withClient().run(client -> diff --git a/client/src/test/java/org/asynchttpclient/RequestBuilderTest.java b/client/src/test/java/org/asynchttpclient/RequestBuilderTest.java index 8ffb9494f7..41fed53a4c 100644 --- a/client/src/test/java/org/asynchttpclient/RequestBuilderTest.java +++ b/client/src/test/java/org/asynchttpclient/RequestBuilderTest.java @@ -20,10 +20,7 @@ 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 void testSettingQueryParamsBeforeUrlShouldNotProduceNPE() { Request request = requestBuilder.build(); assertEquals(request.getUrl(), "http://localhost?key=value"); } + + @Test + public void testSettingHeadersUsingMapWithStringKeys() { + Map> 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"); + } } diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/NonBlockingSemaphoreTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/NonBlockingSemaphoreTest.java deleted file mode 100644 index a387ba408b..0000000000 --- a/client/src/test/java/org/asynchttpclient/netty/channel/NonBlockingSemaphoreTest.java +++ /dev/null @@ -1,76 +0,0 @@ -/* - * Copyright (c) 2017 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.netty.channel; - -import org.testng.annotations.Test; - -import java.util.concurrent.Semaphore; - -import static org.testng.Assert.*; - -/** - * @author Stepan Koltsov - */ -public class NonBlockingSemaphoreTest { - - @Test - public void test0() { - Mirror mirror = new Mirror(0); - assertFalse(mirror.tryAcquire()); - } - - @Test - public void three() { - Mirror mirror = new Mirror(3); - for (int i = 0; i < 3; ++i) { - assertTrue(mirror.tryAcquire()); - } - assertFalse(mirror.tryAcquire()); - mirror.release(); - assertTrue(mirror.tryAcquire()); - } - - @Test - public void negative() { - Mirror mirror = new Mirror(-1); - assertFalse(mirror.tryAcquire()); - mirror.release(); - assertFalse(mirror.tryAcquire()); - mirror.release(); - assertTrue(mirror.tryAcquire()); - } - - private static class Mirror { - private final Semaphore real; - private final NonBlockingSemaphore nonBlocking; - - Mirror(int permits) { - real = new Semaphore(permits); - nonBlocking = new NonBlockingSemaphore(permits); - } - - boolean tryAcquire() { - boolean a = real.tryAcquire(); - boolean b = nonBlocking.tryAcquire(); - assertEquals(a, b); - return a; - } - - void release() { - real.release(); - nonBlocking.release(); - } - } - -} diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/SemaphoreRunner.java b/client/src/test/java/org/asynchttpclient/netty/channel/SemaphoreRunner.java new file mode 100644 index 0000000000..7bff799ceb --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/channel/SemaphoreRunner.java @@ -0,0 +1,52 @@ +package org.asynchttpclient.netty.channel; + +class SemaphoreRunner { + + final ConnectionSemaphore semaphore; + final Thread acquireThread; + + volatile long acquireTime; + volatile Exception acquireException; + + public SemaphoreRunner(ConnectionSemaphore semaphore, Object partitionKey) { + this.semaphore = semaphore; + this.acquireThread = new Thread(() -> { + long beforeAcquire = System.currentTimeMillis(); + try { + semaphore.acquireChannelLock(partitionKey); + } catch (Exception e) { + acquireException = e; + } finally { + acquireTime = System.currentTimeMillis() - beforeAcquire; + } + }); + } + + public void acquire() { + this.acquireThread.start(); + } + + public void interrupt() { + this.acquireThread.interrupt(); + } + + public void await() { + try { + this.acquireThread.join(); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + + public boolean finished() { + return !this.acquireThread.isAlive(); + } + + public long getAcquireTime() { + return acquireTime; + } + + public Exception getAcquireException() { + return acquireException; + } +} diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/SemaphoreTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/SemaphoreTest.java new file mode 100644 index 0000000000..125cd9b066 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/channel/SemaphoreTest.java @@ -0,0 +1,143 @@ +package org.asynchttpclient.netty.channel; + +import org.asynchttpclient.exception.TooManyConnectionsException; +import org.asynchttpclient.exception.TooManyConnectionsPerHostException; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +import java.io.IOException; +import java.util.List; +import java.util.Objects; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import static org.testng.AssertJUnit.*; + +public class SemaphoreTest { + + static final int CHECK_ACQUIRE_TIME__PERMITS = 10; + static final int CHECK_ACQUIRE_TIME__TIMEOUT = 100; + + static final int NON_DETERMINISTIC__INVOCATION_COUNT = 10; + static final int NON_DETERMINISTIC__SUCCESS_PERCENT = 70; + + private final Object PK = new Object(); + + @DataProvider(name = "permitsAndRunnersCount") + public Object[][] permitsAndRunnersCount() { + Object[][] objects = new Object[100][]; + int row = 0; + for (int i = 0; i < 10; i++) { + for (int j = 0; j < 10; j++) { + objects[row++] = new Object[]{i, j}; + } + } + return objects; + } + + @Test(timeOut = 1000, dataProvider = "permitsAndRunnersCount") + public void maxConnectionCheckPermitCount(int permitCount, int runnerCount) { + allSemaphoresCheckPermitCount(new MaxConnectionSemaphore(permitCount, 0), permitCount, runnerCount); + } + + @Test(timeOut = 1000, dataProvider = "permitsAndRunnersCount") + public void perHostCheckPermitCount(int permitCount, int runnerCount) { + allSemaphoresCheckPermitCount(new PerHostConnectionSemaphore(permitCount, 0), permitCount, runnerCount); + } + + @Test(timeOut = 3000, dataProvider = "permitsAndRunnersCount") + public void combinedCheckPermitCount(int permitCount, int runnerCount) { + allSemaphoresCheckPermitCount(new CombinedConnectionSemaphore(permitCount, permitCount, 0), permitCount, runnerCount); + allSemaphoresCheckPermitCount(new CombinedConnectionSemaphore(0, permitCount, 0), permitCount, runnerCount); + allSemaphoresCheckPermitCount(new CombinedConnectionSemaphore(permitCount, 0, 0), permitCount, runnerCount); + } + + private void allSemaphoresCheckPermitCount(ConnectionSemaphore semaphore, int permitCount, int runnerCount) { + List runners = IntStream.range(0, runnerCount) + .mapToObj(i -> new SemaphoreRunner(semaphore, PK)) + .collect(Collectors.toList()); + runners.forEach(SemaphoreRunner::acquire); + runners.forEach(SemaphoreRunner::await); + + long tooManyConnectionsCount = runners.stream().map(SemaphoreRunner::getAcquireException) + .filter(Objects::nonNull) + .filter(e -> e instanceof IOException) + .count(); + + long acquired = runners.stream().map(SemaphoreRunner::getAcquireException) + .filter(Objects::isNull) + .count(); + + int expectedAcquired = permitCount > 0 ? Math.min(permitCount, runnerCount) : runnerCount; + + assertEquals(expectedAcquired, acquired); + assertEquals(runnerCount - acquired, tooManyConnectionsCount); + } + + @Test(timeOut = 1000, invocationCount = NON_DETERMINISTIC__INVOCATION_COUNT, successPercentage = NON_DETERMINISTIC__SUCCESS_PERCENT) + public void maxConnectionCheckAcquireTime() { + checkAcquireTime(new MaxConnectionSemaphore(CHECK_ACQUIRE_TIME__PERMITS, CHECK_ACQUIRE_TIME__TIMEOUT)); + } + + @Test(timeOut = 1000, invocationCount = NON_DETERMINISTIC__INVOCATION_COUNT, successPercentage = NON_DETERMINISTIC__SUCCESS_PERCENT) + public void perHostCheckAcquireTime() { + checkAcquireTime(new PerHostConnectionSemaphore(CHECK_ACQUIRE_TIME__PERMITS, CHECK_ACQUIRE_TIME__TIMEOUT)); + } + + @Test(timeOut = 1000, invocationCount = NON_DETERMINISTIC__INVOCATION_COUNT, successPercentage = NON_DETERMINISTIC__SUCCESS_PERCENT) + public void combinedCheckAcquireTime() { + checkAcquireTime(new CombinedConnectionSemaphore(CHECK_ACQUIRE_TIME__PERMITS, + CHECK_ACQUIRE_TIME__PERMITS, + CHECK_ACQUIRE_TIME__TIMEOUT)); + } + + private void checkAcquireTime(ConnectionSemaphore semaphore) { + List runners = IntStream.range(0, CHECK_ACQUIRE_TIME__PERMITS * 2) + .mapToObj(i -> new SemaphoreRunner(semaphore, PK)) + .collect(Collectors.toList()); + long acquireStartTime = System.currentTimeMillis(); + runners.forEach(SemaphoreRunner::acquire); + runners.forEach(SemaphoreRunner::await); + long timeToAcquire = System.currentTimeMillis() - acquireStartTime; + + assertTrue("Semaphore acquired too soon: " + timeToAcquire+" ms",timeToAcquire >= (CHECK_ACQUIRE_TIME__TIMEOUT - 50)); //Lower Bound + assertTrue("Semaphore acquired too late: " + timeToAcquire+" ms",timeToAcquire <= (CHECK_ACQUIRE_TIME__TIMEOUT + 300)); //Upper Bound + } + + @Test(timeOut = 1000) + public void maxConnectionCheckRelease() throws IOException { + checkRelease(new MaxConnectionSemaphore(1, 0)); + } + + @Test(timeOut = 1000) + public void perHostCheckRelease() throws IOException { + checkRelease(new PerHostConnectionSemaphore(1, 0)); + } + + @Test(timeOut = 1000) + public void combinedCheckRelease() throws IOException { + checkRelease(new CombinedConnectionSemaphore(1, 1, 0)); + } + + private void checkRelease(ConnectionSemaphore semaphore) throws IOException { + semaphore.acquireChannelLock(PK); + boolean tooManyCaught = false; + try { + semaphore.acquireChannelLock(PK); + } catch (TooManyConnectionsException | TooManyConnectionsPerHostException e) { + tooManyCaught = true; + } + assertTrue(tooManyCaught); + tooManyCaught = false; + semaphore.releaseChannelLock(PK); + try { + semaphore.acquireChannelLock(PK); + } catch (TooManyConnectionsException | TooManyConnectionsPerHostException e) { + tooManyCaught = true; + } + assertFalse(tooManyCaught); + } + + +} + diff --git a/client/src/test/java/org/asynchttpclient/proxy/HttpsProxyTest.java b/client/src/test/java/org/asynchttpclient/proxy/HttpsProxyTest.java index 481767022e..a8a1e8d3d3 100644 --- a/client/src/test/java/org/asynchttpclient/proxy/HttpsProxyTest.java +++ b/client/src/test/java/org/asynchttpclient/proxy/HttpsProxyTest.java @@ -13,6 +13,7 @@ package org.asynchttpclient.proxy; import org.asynchttpclient.*; +import org.asynchttpclient.request.body.generator.ByteArrayBodyGenerator; import org.asynchttpclient.test.EchoHandler; import org.eclipse.jetty.proxy.ConnectHandler; import org.eclipse.jetty.server.Server; @@ -23,6 +24,7 @@ import org.testng.annotations.Test; import static org.asynchttpclient.Dsl.*; +import static org.asynchttpclient.test.TestUtils.LARGE_IMAGE_BYTES; import static org.asynchttpclient.test.TestUtils.addHttpConnector; import static org.asynchttpclient.test.TestUtils.addHttpsConnector; import static org.testng.Assert.assertEquals; @@ -84,6 +86,36 @@ public void testConfigProxy() throws Exception { } } + @Test + public void testNoDirectRequestBodyWithProxy() throws Exception { + AsyncHttpClientConfig config = config() + .setFollowRedirect(true) + .setProxyServer(proxyServer("localhost", port1).build()) + .setUseInsecureTrustManager(true) + .build(); + try (AsyncHttpClient asyncHttpClient = asyncHttpClient(config)) { + Response r = asyncHttpClient.executeRequest(post(getTargetUrl2()).setBody(new ByteArrayBodyGenerator(LARGE_IMAGE_BYTES))).get(); + assertEquals(r.getStatusCode(), 200); + } + } + + @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 { diff --git a/client/src/test/java/org/asynchttpclient/request/body/InputStreamPartLargeFileTest.java b/client/src/test/java/org/asynchttpclient/request/body/InputStreamPartLargeFileTest.java new file mode 100644 index 0000000000..48d45341b5 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/request/body/InputStreamPartLargeFileTest.java @@ -0,0 +1,104 @@ +/* + * 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); + } + } +} diff --git a/client/src/test/java/org/asynchttpclient/request/body/multipart/MultipartBodyTest.java b/client/src/test/java/org/asynchttpclient/request/body/multipart/MultipartBodyTest.java index 0b6c5fe6f2..fc54d396ac 100644 --- a/client/src/test/java/org/asynchttpclient/request/body/multipart/MultipartBodyTest.java +++ b/client/src/test/java/org/asynchttpclient/request/body/multipart/MultipartBodyTest.java @@ -19,8 +19,7 @@ import org.asynchttpclient.request.body.Body.BodyState; import org.testng.annotations.Test; -import java.io.File; -import java.io.IOException; +import java.io.*; import java.net.URISyntaxException; import java.net.URL; import java.nio.ByteBuffer; @@ -63,7 +62,15 @@ private static File getTestfile() throws URISyntaxException { } private static MultipartBody buildMultipart() { - return MultipartUtils.newMultipartBody(PARTS, EmptyHttpHeaders.INSTANCE); + List parts = new ArrayList<>(PARTS); + try { + File testFile = getTestfile(); + InputStream inputStream = new BufferedInputStream(new FileInputStream(testFile)); + parts.add(new InputStreamPart("isPart", inputStream, testFile.getName(), testFile.length())); + } catch (URISyntaxException | FileNotFoundException e) { + throw new ExceptionInInitializerError(e); + } + return MultipartUtils.newMultipartBody(parts, EmptyHttpHeaders.INSTANCE); } private static long transferWithCopy(MultipartBody multipartBody, int bufferSize) throws IOException { diff --git a/client/src/test/java/org/asynchttpclient/request/body/multipart/MultipartUploadTest.java b/client/src/test/java/org/asynchttpclient/request/body/multipart/MultipartUploadTest.java index 77584ecdf3..879a40a9d7 100644 --- a/client/src/test/java/org/asynchttpclient/request/body/multipart/MultipartUploadTest.java +++ b/client/src/test/java/org/asynchttpclient/request/body/multipart/MultipartUploadTest.java @@ -77,21 +77,33 @@ public void testSendingSmallFilesAndByteArray() throws Exception { File testResource1File = getClasspathFile(testResource1); File testResource2File = getClasspathFile(testResource2); File testResource3File = getClasspathFile(testResource3); + InputStream inputStreamFile1 = new BufferedInputStream(new FileInputStream(testResource1File)); + InputStream inputStreamFile2 = new BufferedInputStream(new FileInputStream(testResource2File)); + InputStream inputStreamFile3 = new BufferedInputStream(new FileInputStream(testResource3File)); List testFiles = new ArrayList<>(); testFiles.add(testResource1File); testFiles.add(testResource2File); testFiles.add(testResource3File); + testFiles.add(testResource3File); + testFiles.add(testResource2File); + testFiles.add(testResource1File); List expected = new ArrayList<>(); expected.add(expectedContents); expected.add(expectedContents2); expected.add(expectedContents3); + expected.add(expectedContents3); + expected.add(expectedContents2); + expected.add(expectedContents); List gzipped = new ArrayList<>(); gzipped.add(false); gzipped.add(true); gzipped.add(false); + gzipped.add(false); + gzipped.add(true); + gzipped.add(false); File tmpFile = File.createTempFile("textbytearray", ".txt"); try (OutputStream os = Files.newOutputStream(tmpFile.toPath())) { @@ -109,8 +121,11 @@ public void testSendingSmallFilesAndByteArray() throws Exception { .addBodyPart(new StringPart("Name", "Dominic")) .addBodyPart(new FilePart("file3", testResource3File, "text/plain", UTF_8)) .addBodyPart(new StringPart("Age", "3")).addBodyPart(new StringPart("Height", "shrimplike")) + .addBodyPart(new InputStreamPart("inputStream3", inputStreamFile3, testResource3File.getName(), testResource3File.length(), "text/plain", UTF_8)) + .addBodyPart(new InputStreamPart("inputStream2", inputStreamFile2, testResource2File.getName(), testResource2File.length(), "application/x-gzip", null)) .addBodyPart(new StringPart("Hair", "ridiculous")).addBodyPart(new ByteArrayPart("file4", expectedContents.getBytes(UTF_8), "text/plain", UTF_8, "bytearray.txt")) + .addBodyPart(new InputStreamPart("inputStream1", inputStreamFile1, testResource1File.getName(), testResource1File.length(), "text/plain", UTF_8)) .build(); Response res = c.executeRequest(r).get(); @@ -142,6 +157,65 @@ public void sendEmptyFileZeroCopy() throws Exception { sendEmptyFile0(false); } + private void sendEmptyFileInputStream(boolean disableZeroCopy) throws Exception { + File file = getClasspathFile("empty.txt"); + try (AsyncHttpClient c = asyncHttpClient(config().setDisableZeroCopy(disableZeroCopy))) { + InputStream inputStream = new BufferedInputStream(new FileInputStream(file)); + Request r = post("http://localhost" + ":" + port1 + "/upload") + .addBodyPart(new InputStreamPart("file", inputStream, file.getName(), file.length(), "text/plain", UTF_8)).build(); + + Response res = c.executeRequest(r).get(); + assertEquals(res.getStatusCode(), 200); + } + } + + @Test + public void testSendEmptyFileInputStream() throws Exception { + sendEmptyFileInputStream(true); + } + + @Test + public void testSendEmptyFileInputStreamZeroCopy() throws Exception { + sendEmptyFileInputStream(false); + } + + private void sendFileInputStream(boolean useContentLength, boolean disableZeroCopy) throws Exception { + File file = getClasspathFile("textfile.txt"); + try (AsyncHttpClient c = asyncHttpClient(config().setDisableZeroCopy(disableZeroCopy))) { + InputStream inputStream = new BufferedInputStream(new FileInputStream(file)); + InputStreamPart part; + if (useContentLength) { + part = new InputStreamPart("file", inputStream, file.getName(), file.length()); + } else { + part = new InputStreamPart("file", inputStream, file.getName()); + } + Request r = post("http://localhost" + ":" + port1 + "/upload").addBodyPart(part).build(); + + Response res = c.executeRequest(r).get(); + assertEquals(res.getStatusCode(), 200); + } + } + + @Test + public void testSendFileInputStreamUnknownContentLength() throws Exception { + sendFileInputStream(false, true); + } + + @Test + public void testSendFileInputStreamZeroCopyUnknownContentLength() throws Exception { + sendFileInputStream(false, false); + } + + @Test + public void testSendFileInputStreamKnownContentLength() throws Exception { + sendFileInputStream(true, true); + } + + @Test + public void testSendFileInputStreamZeroCopyKnownContentLength() throws Exception { + sendFileInputStream(true, false); + } + /** * Test that the files were sent, based on the response from the servlet */ diff --git a/client/src/test/java/org/asynchttpclient/spnego/SpnegoEngineTest.java b/client/src/test/java/org/asynchttpclient/spnego/SpnegoEngineTest.java new file mode 100644 index 0000000000..92ff4a4d78 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/spnego/SpnegoEngineTest.java @@ -0,0 +1,163 @@ +package org.asynchttpclient.spnego; + +import org.apache.commons.io.FileUtils; +import org.apache.kerby.kerberos.kerb.server.SimpleKdcServer; +import org.asynchttpclient.AbstractBasicTest; +import org.testng.Assert; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +import java.io.File; +import java.util.HashMap; +import java.util.Map; + +public class SpnegoEngineTest extends AbstractBasicTest { + private static SimpleKdcServer kerbyServer; + + private static String basedir; + private static String alice; + private static String bob; + private static File aliceKeytab; + private static File bobKeytab; + private static File loginConfig; + + @BeforeClass + public static void startServers() throws Exception { + basedir = System.getProperty("basedir"); + if (basedir == null) { + basedir = new File(".").getCanonicalPath(); + } + + // System.setProperty("sun.security.krb5.debug", "true"); + System.setProperty("java.security.krb5.conf", + new File(basedir + File.separator + "target" + File.separator + "krb5.conf").getCanonicalPath()); + loginConfig = new File(basedir + File.separator + "target" + File.separator + "kerberos.jaas"); + System.setProperty("java.security.auth.login.config", loginConfig.getCanonicalPath()); + + kerbyServer = new SimpleKdcServer(); + + kerbyServer.setKdcRealm("service.ws.apache.org"); + kerbyServer.setAllowUdp(false); + kerbyServer.setWorkDir(new File(basedir, "target")); + + //kerbyServer.setInnerKdcImpl(new NettyKdcServerImpl(kerbyServer.getKdcSetting())); + + kerbyServer.init(); + + // Create principals + alice = "alice@service.ws.apache.org"; + bob = "bob/service.ws.apache.org@service.ws.apache.org"; + + kerbyServer.createPrincipal(alice, "alice"); + kerbyServer.createPrincipal(bob, "bob"); + + aliceKeytab = new File(basedir + File.separator + "target" + File.separator + "alice.keytab"); + bobKeytab = new File(basedir + File.separator + "target" + File.separator + "bob.keytab"); + kerbyServer.exportPrincipal(alice, aliceKeytab); + kerbyServer.exportPrincipal(bob, bobKeytab); + + kerbyServer.start(); + + FileUtils.copyInputStreamToFile(SpnegoEngine.class.getResourceAsStream("/kerberos.jaas"), loginConfig); + } + + @Test + public void testSpnegoGenerateTokenWithUsernamePassword() throws Exception { + SpnegoEngine spnegoEngine = new SpnegoEngine("alice", + "alice", + "bob", + "service.ws.apache.org", + false, + null, + "alice", + null); + String token = spnegoEngine.generateToken("localhost"); + Assert.assertNotNull(token); + Assert.assertTrue(token.startsWith("YII")); + } + + @Test(expectedExceptions = SpnegoEngineException.class) + public void testSpnegoGenerateTokenWithUsernamePasswordFail() throws Exception { + SpnegoEngine spnegoEngine = new SpnegoEngine("alice", + "wrong password", + "bob", + "service.ws.apache.org", + false, + null, + "alice", + null); + spnegoEngine.generateToken("localhost"); + } + + @Test + public void testSpnegoGenerateTokenWithCustomLoginConfig() throws Exception { + Map loginConfig = new HashMap<>(); + loginConfig.put("useKeyTab", "true"); + loginConfig.put("storeKey", "true"); + loginConfig.put("refreshKrb5Config", "true"); + loginConfig.put("keyTab", aliceKeytab.getCanonicalPath()); + loginConfig.put("principal", alice); + loginConfig.put("debug", String.valueOf(true)); + SpnegoEngine spnegoEngine = new SpnegoEngine(null, + null, + "bob", + "service.ws.apache.org", + false, + loginConfig, + null, + null); + + String token = spnegoEngine.generateToken("localhost"); + Assert.assertNotNull(token); + Assert.assertTrue(token.startsWith("YII")); + } + + @Test + public void testGetCompleteServicePrincipalName() throws Exception { + { + SpnegoEngine spnegoEngine = new SpnegoEngine(null, + null, + "bob", + "service.ws.apache.org", + false, + null, + null, + null); + Assert.assertEquals("bob@service.ws.apache.org", spnegoEngine.getCompleteServicePrincipalName("localhost")); + } + { + SpnegoEngine spnegoEngine = new SpnegoEngine(null, + null, + null, + "service.ws.apache.org", + true, + null, + null, + null); + Assert.assertNotEquals("HTTP@localhost", spnegoEngine.getCompleteServicePrincipalName("localhost")); + Assert.assertTrue(spnegoEngine.getCompleteServicePrincipalName("localhost").startsWith("HTTP@")); + } + { + SpnegoEngine spnegoEngine = new SpnegoEngine(null, + null, + null, + "service.ws.apache.org", + false, + null, + null, + null); + Assert.assertEquals("HTTP@localhost", spnegoEngine.getCompleteServicePrincipalName("localhost")); + } + } + + @AfterClass + public static void cleanup() throws Exception { + if (kerbyServer != null) { + kerbyServer.stop(); + } + FileUtils.deleteQuietly(aliceKeytab); + FileUtils.deleteQuietly(bobKeytab); + FileUtils.deleteQuietly(loginConfig); + } +} diff --git a/client/src/test/java/org/asynchttpclient/test/EchoHandler.java b/client/src/test/java/org/asynchttpclient/test/EchoHandler.java index ee19f2ee0a..6826155648 100644 --- a/client/src/test/java/org/asynchttpclient/test/EchoHandler.java +++ b/client/src/test/java/org/asynchttpclient/test/EchoHandler.java @@ -23,11 +23,15 @@ import javax.servlet.http.Cookie; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; +import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.nio.charset.StandardCharsets; import java.util.Enumeration; +import java.util.zip.Deflater; -import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH; -import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_MD5; +import static io.netty.handler.codec.http.HttpHeaderNames.*; +import static io.netty.handler.codec.http.HttpHeaderValues.CHUNKED; +import static io.netty.handler.codec.http.HttpHeaderValues.DEFLATE; public class EchoHandler extends AbstractHandler { @@ -69,6 +73,15 @@ public void handle(String pathInContext, Request request, HttpServletRequest htt httpResponse.sendRedirect(httpRequest.getHeader("X-redirect")); return; } + if (headerName.startsWith("X-fail")) { + byte[] body = "custom error message".getBytes(StandardCharsets.US_ASCII); + httpResponse.addHeader(CONTENT_LENGTH.toString(), String.valueOf(body.length)); + httpResponse.setStatus(HttpServletResponse.SC_EXPECTATION_FAILED); + httpResponse.getOutputStream().write(body); + httpResponse.getOutputStream().flush(); + httpResponse.getOutputStream().close(); + return; + } httpResponse.addHeader("X-" + headerName, httpRequest.getHeader(headerName)); } @@ -105,18 +118,14 @@ public void handle(String pathInContext, Request request, HttpServletRequest htt } } - String requestBodyLength = httpRequest.getHeader("X-" + CONTENT_LENGTH); - - if (requestBodyLength != null) { - byte[] requestBodyBytes = IOUtils.toByteArray(httpRequest.getInputStream()); - int total = requestBodyBytes.length; + if (httpRequest.getHeader("X-COMPRESS") != null) { + byte[] compressed = deflate(IOUtils.toByteArray(httpRequest.getInputStream())); + httpResponse.addIntHeader(CONTENT_LENGTH.toString(), compressed.length); + httpResponse.addHeader(CONTENT_ENCODING.toString(), DEFLATE.toString()); + httpResponse.getOutputStream().write(compressed, 0, compressed.length); - httpResponse.addIntHeader("X-" + CONTENT_LENGTH, total); - String md5 = TestUtils.md5(requestBodyBytes, 0, total); - httpResponse.addHeader(CONTENT_MD5.toString(), md5); - - httpResponse.getOutputStream().write(requestBodyBytes, 0, total); } else { + httpResponse.addHeader(TRANSFER_ENCODING.toString(), CHUNKED.toString()); int size = 16384; if (httpRequest.getContentLength() > 0) { size = httpRequest.getContentLength(); @@ -138,4 +147,21 @@ public void handle(String pathInContext, Request request, HttpServletRequest htt // FIXME don't always close, depends on the test, cf ReactiveStreamsTest httpResponse.getOutputStream().close(); } + + private static byte[] deflate(byte[] input) throws IOException { + Deflater compressor = new Deflater(); + compressor.setLevel(Deflater.BEST_COMPRESSION); + + compressor.setInput(input); + compressor.finish(); + + try (ByteArrayOutputStream bos = new ByteArrayOutputStream(input.length)) { + byte[] buf = new byte[1024]; + while (!compressor.finished()) { + int count = compressor.deflate(buf); + bos.write(buf, 0, count); + } + return bos.toByteArray(); + } + } } diff --git a/client/src/test/java/org/asynchttpclient/test/EventCollectingHandler.java b/client/src/test/java/org/asynchttpclient/test/EventCollectingHandler.java index 252de41913..8047c5f843 100644 --- a/client/src/test/java/org/asynchttpclient/test/EventCollectingHandler.java +++ b/client/src/test/java/org/asynchttpclient/test/EventCollectingHandler.java @@ -20,6 +20,7 @@ import org.asynchttpclient.netty.request.NettyRequest; import org.testng.Assert; +import javax.net.ssl.SSLSession; import java.net.InetSocketAddress; import java.util.List; import java.util.Queue; @@ -128,7 +129,8 @@ public void onTlsHandshakeAttempt() { } @Override - public void onTlsHandshakeSuccess() { + public void onTlsHandshakeSuccess(SSLSession sslSession) { + Assert.assertNotNull(sslSession); firedEvents.add(TLS_HANDSHAKE_SUCCESS_EVENT); } diff --git a/client/src/test/java/org/asynchttpclient/testserver/HttpServer.java b/client/src/test/java/org/asynchttpclient/testserver/HttpServer.java index ae387469a7..9b74656b5e 100644 --- a/client/src/test/java/org/asynchttpclient/testserver/HttpServer.java +++ b/client/src/test/java/org/asynchttpclient/testserver/HttpServer.java @@ -25,6 +25,8 @@ import javax.servlet.http.HttpServletResponse; import java.io.Closeable; import java.io.IOException; +import java.net.URLEncoder; +import java.nio.charset.StandardCharsets; import java.util.Enumeration; import java.util.Map.Entry; import java.util.concurrent.ConcurrentLinkedQueue; @@ -208,8 +210,9 @@ protected void handle0(String target, Request baseRequest, HttpServletRequest re response.addHeader("X-" + headerName, request.getHeader(headerName)); } + StringBuilder requestBody = new StringBuilder(); for (Entry e : baseRequest.getParameterMap().entrySet()) { - response.addHeader("X-" + e.getKey(), e.getValue()[0]); + response.addHeader("X-" + e.getKey(), URLEncoder.encode(e.getValue()[0], StandardCharsets.UTF_8.name())); } Cookie[] cs = request.getCookies(); @@ -219,14 +222,6 @@ protected void handle0(String target, Request baseRequest, HttpServletRequest re } } - Enumeration parameterNames = request.getParameterNames(); - StringBuilder requestBody = new StringBuilder(); - while (parameterNames.hasMoreElements()) { - String param = parameterNames.nextElement(); - response.addHeader("X-" + param, request.getParameter(param)); - requestBody.append(param); - requestBody.append("_"); - } if (requestBody.length() > 0) { response.getOutputStream().write(requestBody.toString().getBytes()); } diff --git a/client/src/test/resources/kerberos.jaas b/client/src/test/resources/kerberos.jaas new file mode 100644 index 0000000000..cd5b316bf1 --- /dev/null +++ b/client/src/test/resources/kerberos.jaas @@ -0,0 +1,8 @@ + +alice { + com.sun.security.auth.module.Krb5LoginModule required refreshKrb5Config=true useKeyTab=false principal="alice"; +}; + +bob { + com.sun.security.auth.module.Krb5LoginModule required refreshKrb5Config=true useKeyTab=false storeKey=true principal="bob/service.ws.apache.org"; +}; diff --git a/extras/pom.xml b/extras/pom.xml index b2f1c843f0..01bd9b7879 100644 --- a/extras/pom.xml +++ b/extras/pom.xml @@ -29,12 +29,5 @@ async-http-client ${project.version} - - org.asynchttpclient - async-http-client - 2.5.3-SNAPSHOT - test - tests - diff --git a/extras/retrofit2/pom.xml b/extras/retrofit2/pom.xml index bd286c70d8..f86e154949 100644 --- a/extras/retrofit2/pom.xml +++ b/extras/retrofit2/pom.xml @@ -12,7 +12,7 @@ The Async Http Client Retrofit2 Extras. - 2.4.0 + 2.5.0 1.18.6 diff --git a/extras/retrofit2/src/main/java/org/asynchttpclient/extras/retrofit/AsyncHttpClientCall.java b/extras/retrofit2/src/main/java/org/asynchttpclient/extras/retrofit/AsyncHttpClientCall.java index e8980fddab..bbd7601873 100644 --- a/extras/retrofit2/src/main/java/org/asynchttpclient/extras/retrofit/AsyncHttpClientCall.java +++ b/extras/retrofit2/src/main/java/org/asynchttpclient/extras/retrofit/AsyncHttpClientCall.java @@ -17,6 +17,7 @@ import lombok.extern.slf4j.Slf4j; import okhttp3.*; import okio.Buffer; +import okio.Timeout; import org.asynchttpclient.AsyncCompletionHandler; import org.asynchttpclient.AsyncHttpClient; import org.asynchttpclient.RequestBuilder; @@ -29,6 +30,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; +import java.util.function.Supplier; /** * {@link AsyncHttpClient} Retrofit2 {@link okhttp3.Call} @@ -38,13 +40,8 @@ @Builder(toBuilder = true) @Slf4j class AsyncHttpClientCall implements Cloneable, okhttp3.Call { - /** - * Default {@link #execute()} timeout in milliseconds (value: {@value}) - * - * @see #execute() - * @see #executeTimeoutMillis - */ - public static final long DEFAULT_EXECUTE_TIMEOUT_MILLIS = 30_000; + private static final ResponseBody EMPTY_BODY = ResponseBody.create(null, ""); + /** * Tells whether call has been executed. * @@ -52,37 +49,38 @@ class AsyncHttpClientCall implements Cloneable, okhttp3.Call { * @see #isCanceled() */ private final AtomicReference> futureRef = new AtomicReference<>(); + /** - * HttpClient instance. + * {@link AsyncHttpClient} supplier */ @NonNull - AsyncHttpClient httpClient; - /** - * {@link #execute()} response timeout in milliseconds. - */ - @Builder.Default - long executeTimeoutMillis = DEFAULT_EXECUTE_TIMEOUT_MILLIS; + Supplier httpClientSupplier; + /** * Retrofit request. */ @NonNull @Getter(AccessLevel.NONE) Request request; + /** * List of consumers that get called just before actual async-http-client request is being built. */ @Singular("requestCustomizer") List> requestCustomizers; + /** * List of consumers that get called just before actual HTTP request is being fired. */ @Singular("onRequestStart") List> onRequestStart; + /** * List of consumers that get called when HTTP request finishes with an exception. */ @Singular("onRequestFailure") List> onRequestFailure; + /** * List of consumers that get called when HTTP request finishes successfully. */ @@ -128,7 +126,7 @@ public Request request() { @Override public Response execute() throws IOException { try { - return executeHttpRequest().get(getExecuteTimeoutMillis(), TimeUnit.MILLISECONDS); + return executeHttpRequest().get(getRequestTimeoutMillis(), TimeUnit.MILLISECONDS); } catch (ExecutionException e) { throw toIOException(e.getCause()); } catch (Exception e) { @@ -146,7 +144,7 @@ public void enqueue(Callback responseCallback) { @Override public void cancel() { val future = futureRef.get(); - if (future != null) { + if (future != null && !future.isDone()) { if (!future.cancel(true)) { log.warn("Cannot cancel future: {}", future); } @@ -165,6 +163,20 @@ public boolean isCanceled() { return future != null && future.isCancelled(); } + @Override + public Timeout timeout() { + return new Timeout().timeout(getRequestTimeoutMillis(), TimeUnit.MILLISECONDS); + } + + /** + * Returns HTTP request timeout in milliseconds, retrieved from http client configuration. + * + * @return request timeout in milliseconds. + */ + protected long getRequestTimeoutMillis() { + return Math.abs(getHttpClient().getConfig().getRequestTimeout()); + } + @Override public Call clone() { return toBuilder().build(); @@ -228,6 +240,20 @@ public Response onCompleted(org.asynchttpclient.Response response) { return future; } + /** + * Returns HTTP client. + * + * @return http client + * @throws IllegalArgumentException if {@link #httpClientSupplier} returned {@code null}. + */ + protected AsyncHttpClient getHttpClient() { + val httpClient = httpClientSupplier.get(); + if (httpClient == null) { + throw new IllegalStateException("Async HTTP client instance supplier " + httpClientSupplier + " returned null."); + } + return httpClient; + } + /** * Converts async-http-client response to okhttp response. * @@ -254,6 +280,8 @@ private Response toOkhttpResponse(org.asynchttpclient.Response asyncHttpClientRe ? null : MediaType.parse(asyncHttpClientResponse.getContentType()); val okHttpBody = ResponseBody.create(contentType, asyncHttpClientResponse.getResponseBodyAsBytes()); rspBuilder.body(okHttpBody); + } else { + rspBuilder.body(EMPTY_BODY); } return rspBuilder.build(); diff --git a/extras/retrofit2/src/main/java/org/asynchttpclient/extras/retrofit/AsyncHttpClientCallFactory.java b/extras/retrofit2/src/main/java/org/asynchttpclient/extras/retrofit/AsyncHttpClientCallFactory.java index b7c087fd35..0077cd32e3 100644 --- a/extras/retrofit2/src/main/java/org/asynchttpclient/extras/retrofit/AsyncHttpClientCallFactory.java +++ b/extras/retrofit2/src/main/java/org/asynchttpclient/extras/retrofit/AsyncHttpClientCallFactory.java @@ -19,31 +19,35 @@ import java.util.List; import java.util.function.Consumer; +import java.util.function.Supplier; import static org.asynchttpclient.extras.retrofit.AsyncHttpClientCall.runConsumers; /** - * {@link AsyncHttpClient} implementation of Retrofit2 {@link Call.Factory} + * {@link AsyncHttpClient} implementation of Retrofit2 + * {@link Call.Factory}. */ @Value @Builder(toBuilder = true) public class AsyncHttpClientCallFactory implements Call.Factory { /** - * {@link AsyncHttpClient} in use. + * Supplier of {@link AsyncHttpClient}. */ @NonNull - AsyncHttpClient httpClient; + @Getter(AccessLevel.NONE) + Supplier httpClientSupplier; /** * List of {@link Call} builder customizers that are invoked just before creating it. */ @Singular("callCustomizer") + @Getter(AccessLevel.PACKAGE) List> callCustomizers; @Override public Call newCall(Request request) { val callBuilder = AsyncHttpClientCall.builder() - .httpClient(httpClient) + .httpClientSupplier(httpClientSupplier) .request(request); // customize builder before creating a call @@ -52,4 +56,35 @@ public Call newCall(Request request) { // create a call return callBuilder.build(); } -} + + /** + * Returns {@link AsyncHttpClient} from {@link #httpClientSupplier}. + * + * @return http client. + */ + AsyncHttpClient getHttpClient() { + return httpClientSupplier.get(); + } + + /** + * Builder for {@link AsyncHttpClientCallFactory}. + */ + public static class AsyncHttpClientCallFactoryBuilder { + /** + * {@link AsyncHttpClient} supplier that returns http client to be used to execute HTTP requests. + */ + private Supplier httpClientSupplier; + + /** + * Sets concrete http client to be used by the factory to execute HTTP requests. Invocation of this method + * overrides any previous http client supplier set by {@link #httpClientSupplier(Supplier)}! + * + * @param httpClient http client + * @return reference to itself. + * @see #httpClientSupplier(Supplier) + */ + public AsyncHttpClientCallFactoryBuilder httpClient(@NonNull AsyncHttpClient httpClient) { + return httpClientSupplier(() -> httpClient); + } + } +} \ No newline at end of file diff --git a/extras/retrofit2/src/test/java/org/asynchttpclient/extras/retrofit/AsyncHttpClientCallFactoryTest.java b/extras/retrofit2/src/test/java/org/asynchttpclient/extras/retrofit/AsyncHttpClientCallFactoryTest.java index 730ab5d007..864931a583 100644 --- a/extras/retrofit2/src/test/java/org/asynchttpclient/extras/retrofit/AsyncHttpClientCallFactoryTest.java +++ b/extras/retrofit2/src/test/java/org/asynchttpclient/extras/retrofit/AsyncHttpClientCallFactoryTest.java @@ -14,23 +14,34 @@ import lombok.extern.slf4j.Slf4j; import lombok.val; +import okhttp3.MediaType; import okhttp3.Request; +import okhttp3.RequestBody; import okhttp3.Response; import org.asynchttpclient.AsyncHttpClient; import org.asynchttpclient.RequestBuilder; import org.testng.annotations.Test; +import java.util.Objects; import java.util.UUID; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; -import static org.asynchttpclient.extras.retrofit.AsyncHttpClientCallTest.REQUEST; import static org.asynchttpclient.extras.retrofit.AsyncHttpClientCallTest.createConsumer; import static org.mockito.Mockito.mock; import static org.testng.Assert.*; @Slf4j public class AsyncHttpClientCallFactoryTest { + private static final MediaType MEDIA_TYPE = MediaType.parse("application/json"); + private static final String JSON_BODY = "{\"foo\": \"bar\"}"; + private static final RequestBody BODY = RequestBody.create(MEDIA_TYPE, JSON_BODY); + private static final String URL = "http://localhost:11000/foo/bar?a=b&c=d"; + private static final Request REQUEST = new Request.Builder() + .post(BODY) + .addHeader("X-Foo", "Bar") + .url(URL) + .build(); @Test void newCallShouldProduceExpectedResult() { // given @@ -109,12 +120,12 @@ void shouldApplyAllConsumersToCallBeingConstructed() { }; Consumer callCustomizer = callBuilder -> - callBuilder - .requestCustomizer(requestCustomizer) - .requestCustomizer(rb -> log.warn("I'm customizing: {}", rb)) - .onRequestSuccess(createConsumer(numRequestSuccess)) - .onRequestFailure(createConsumer(numRequestFailure)) - .onRequestStart(createConsumer(numRequestStart)); + callBuilder + .requestCustomizer(requestCustomizer) + .requestCustomizer(rb -> log.warn("I'm customizing: {}", rb)) + .onRequestSuccess(createConsumer(numRequestSuccess)) + .onRequestFailure(createConsumer(numRequestFailure)) + .onRequestStart(createConsumer(numRequestStart)); // create factory val factory = AsyncHttpClientCallFactory.builder() @@ -151,4 +162,65 @@ void shouldApplyAllConsumersToCallBeingConstructed() { assertNotNull(call.getRequestCustomizers()); assertTrue(call.getRequestCustomizers().size() == 2); } + + @Test(expectedExceptions = NullPointerException.class, + expectedExceptionsMessageRegExp = "httpClientSupplier is marked @NonNull but is null") + void shouldThrowISEIfHttpClientIsNotDefined() { + // given + val factory = AsyncHttpClientCallFactory.builder() + .build(); + + // when + val httpClient = factory.getHttpClient(); + + // then + assertNull(httpClient); + } + + @Test + void shouldUseHttpClientInstanceIfSupplierIsNotAvailable() { + // given + val httpClient = mock(AsyncHttpClient.class); + + val factory = AsyncHttpClientCallFactory.builder() + .httpClient(httpClient) + .build(); + + // when + val usedHttpClient = factory.getHttpClient(); + + // then + assertTrue(usedHttpClient == httpClient); + + // when + val call = (AsyncHttpClientCall) factory.newCall(REQUEST); + + // then: call should contain correct http client + assertTrue(call.getHttpClient()== httpClient); + } + + @Test + void shouldPreferHttpClientSupplierOverHttpClient() { + // given + val httpClientA = mock(AsyncHttpClient.class); + val httpClientB = mock(AsyncHttpClient.class); + + val factory = AsyncHttpClientCallFactory.builder() + .httpClient(httpClientA) + .httpClientSupplier(() -> httpClientB) + .build(); + + // when + val usedHttpClient = factory.getHttpClient(); + + // then + assertTrue(usedHttpClient == httpClientB); + + // when: try to create new call + val call = (AsyncHttpClientCall) factory.newCall(REQUEST); + + // then: call should contain correct http client + assertNotNull(call); + assertTrue(call.getHttpClient() == httpClientB); + } } diff --git a/extras/retrofit2/src/test/java/org/asynchttpclient/extras/retrofit/AsyncHttpClientCallTest.java b/extras/retrofit2/src/test/java/org/asynchttpclient/extras/retrofit/AsyncHttpClientCallTest.java index 90ab4b33e5..e655ed73fc 100644 --- a/extras/retrofit2/src/test/java/org/asynchttpclient/extras/retrofit/AsyncHttpClientCallTest.java +++ b/extras/retrofit2/src/test/java/org/asynchttpclient/extras/retrofit/AsyncHttpClientCallTest.java @@ -14,16 +14,18 @@ import io.netty.handler.codec.http.DefaultHttpHeaders; import io.netty.handler.codec.http.EmptyHttpHeaders; -import lombok.val; -import okhttp3.MediaType; -import okhttp3.Request; -import okhttp3.RequestBody; +import lombok.*; +import lombok.extern.slf4j.Slf4j; +import okhttp3.*; import org.asynchttpclient.AsyncCompletionHandler; import org.asynchttpclient.AsyncHttpClient; +import org.asynchttpclient.AsyncHttpClientConfig; import org.asynchttpclient.BoundRequestBuilder; +import org.asynchttpclient.DefaultAsyncHttpClientConfig; import org.asynchttpclient.Response; import org.mockito.ArgumentCaptor; import org.testng.Assert; +import org.testng.annotations.BeforeMethod; import org.testng.annotations.DataProvider; import org.testng.annotations.Test; @@ -35,16 +37,32 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; +import java.util.function.Supplier; +import static java.util.concurrent.TimeUnit.SECONDS; import static org.asynchttpclient.extras.retrofit.AsyncHttpClientCall.runConsumer; import static org.asynchttpclient.extras.retrofit.AsyncHttpClientCall.runConsumers; -import static org.mockito.Matchers.any; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.*; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotEquals; import static org.testng.Assert.assertTrue; +@Slf4j public class AsyncHttpClientCallTest { static final Request REQUEST = new Request.Builder().url("http://www.google.com/").build(); + static final DefaultAsyncHttpClientConfig DEFAULT_AHC_CONFIG = new DefaultAsyncHttpClientConfig.Builder() + .setRequestTimeout(1_000) + .build(); + + private AsyncHttpClient httpClient; + private Supplier httpClientSupplier = () -> httpClient; + + @BeforeMethod + void setup() { + httpClient = mock(AsyncHttpClient.class); + when(httpClient.getConfig()).thenReturn(DEFAULT_AHC_CONFIG); + } @Test(expectedExceptions = NullPointerException.class, dataProvider = "first") void builderShouldThrowInCaseOfMissingProperties(AsyncHttpClientCall.AsyncHttpClientCallBuilder builder) { @@ -53,12 +71,10 @@ void builderShouldThrowInCaseOfMissingProperties(AsyncHttpClientCall.AsyncHttpCl @DataProvider(name = "first") Object[][] dataProviderFirst() { - val httpClient = mock(AsyncHttpClient.class); - return new Object[][]{ {AsyncHttpClientCall.builder()}, {AsyncHttpClientCall.builder().request(REQUEST)}, - {AsyncHttpClientCall.builder().httpClient(httpClient)} + {AsyncHttpClientCall.builder().httpClientSupplier(httpClientSupplier)} }; } @@ -76,7 +92,7 @@ void shouldInvokeConsumersOnEachExecution(Consumer> ha val numRequestCustomizer = new AtomicInteger(); // prepare http client mock - val httpClient = mock(AsyncHttpClient.class); + this.httpClient = mock(AsyncHttpClient.class); val mockRequest = mock(org.asynchttpclient.Request.class); when(mockRequest.getHeaders()).thenReturn(EmptyHttpHeaders.INSTANCE); @@ -93,13 +109,12 @@ void shouldInvokeConsumersOnEachExecution(Consumer> ha // create call instance val call = AsyncHttpClientCall.builder() - .httpClient(httpClient) + .httpClientSupplier(httpClientSupplier) .request(REQUEST) .onRequestStart(e -> numStarted.incrementAndGet()) .onRequestFailure(t -> numFailed.incrementAndGet()) .onRequestSuccess(r -> numOk.incrementAndGet()) .requestCustomizer(rb -> numRequestCustomizer.incrementAndGet()) - .executeTimeoutMillis(1000) .build(); // when @@ -162,7 +177,7 @@ Object[][] dataProviderSecond() { void toIOExceptionShouldProduceExpectedResult(Throwable exception) { // given val call = AsyncHttpClientCall.builder() - .httpClient(mock(AsyncHttpClient.class)) + .httpClientSupplier(httpClientSupplier) .request(REQUEST) .build(); @@ -236,13 +251,12 @@ public void contentTypeHeaderIsPassedInRequest() throws Exception { Request request = requestWithBody(); ArgumentCaptor capture = ArgumentCaptor.forClass(org.asynchttpclient.Request.class); - AsyncHttpClient client = mock(AsyncHttpClient.class); - givenResponseIsProduced(client, aResponse()); + givenResponseIsProduced(httpClient, aResponse()); - whenRequestIsMade(client, request); + whenRequestIsMade(httpClient, request); - verify(client).executeRequest(capture.capture(), any()); + verify(httpClient).executeRequest(capture.capture(), any()); org.asynchttpclient.Request ahcRequest = capture.getValue(); @@ -254,11 +268,9 @@ public void contentTypeHeaderIsPassedInRequest() throws Exception { @Test public void contenTypeIsOptionalInResponse() throws Exception { - AsyncHttpClient client = mock(AsyncHttpClient.class); - - givenResponseIsProduced(client, responseWithBody(null, "test")); + givenResponseIsProduced(httpClient, responseWithBody(null, "test")); - okhttp3.Response response = whenRequestIsMade(client, REQUEST); + okhttp3.Response response = whenRequestIsMade(httpClient, REQUEST); assertEquals(response.code(), 200); assertEquals(response.header("Server"), "nginx"); @@ -268,11 +280,9 @@ public void contenTypeIsOptionalInResponse() throws Exception { @Test public void contentTypeIsProperlyParsedIfPresent() throws Exception { - AsyncHttpClient client = mock(AsyncHttpClient.class); + givenResponseIsProduced(httpClient, responseWithBody("text/plain", "test")); - givenResponseIsProduced(client, responseWithBody("text/plain", "test")); - - okhttp3.Response response = whenRequestIsMade(client, REQUEST); + okhttp3.Response response = whenRequestIsMade(httpClient, REQUEST); assertEquals(response.code(), 200); assertEquals(response.header("Server"), "nginx"); @@ -281,6 +291,73 @@ public void contentTypeIsProperlyParsedIfPresent() throws Exception { } + @Test + public void bodyIsNotNullInResponse() throws Exception { + givenResponseIsProduced(httpClient, responseWithNoBody()); + + okhttp3.Response response = whenRequestIsMade(httpClient, REQUEST); + + assertEquals(response.code(), 200); + assertEquals(response.header("Server"), "nginx"); + assertNotEquals(response.body(), null); + } + + @Test(expectedExceptions = IllegalStateException.class, expectedExceptionsMessageRegExp = ".*returned null.") + void getHttpClientShouldThrowISEIfSupplierReturnsNull() { + // given: + val call = AsyncHttpClientCall.builder() + .httpClientSupplier(() -> null) + .request(requestWithBody()) + .build(); + + // when: should throw ISE + call.getHttpClient(); + } + + @Test + void shouldReturnTimeoutSpecifiedInAHCInstanceConfig() { + // given: + val cfgBuilder = new DefaultAsyncHttpClientConfig.Builder(); + AsyncHttpClientConfig config = null; + + // and: setup call + val call = AsyncHttpClientCall.builder() + .httpClientSupplier(httpClientSupplier) + .request(requestWithBody()) + .build(); + + // when: set read timeout to 5s, req timeout to 6s + config = cfgBuilder.setReadTimeout((int) SECONDS.toMillis(5)) + .setRequestTimeout((int) SECONDS.toMillis(6)) + .build(); + when(httpClient.getConfig()).thenReturn(config); + + // then: expect request timeout + assertEquals(call.getRequestTimeoutMillis(), SECONDS.toMillis(6)); + assertEquals(call.timeout().timeoutNanos(), SECONDS.toNanos(6)); + + // when: set read timeout to 10 seconds, req timeout to 7s + config = cfgBuilder.setReadTimeout((int) SECONDS.toMillis(10)) + .setRequestTimeout((int) SECONDS.toMillis(7)) + .build(); + when(httpClient.getConfig()).thenReturn(config); + + // then: expect request timeout + assertEquals(call.getRequestTimeoutMillis(), SECONDS.toMillis(7)); + assertEquals(call.timeout().timeoutNanos(), SECONDS.toNanos(7)); + + // when: set request timeout to a negative value, just for fun. + config = cfgBuilder.setRequestTimeout(-1000) + .setReadTimeout(2000) + .build(); + + when(httpClient.getConfig()).thenReturn(config); + + // then: expect request timeout, but as positive value + assertEquals(call.getRequestTimeoutMillis(), SECONDS.toMillis(1)); + assertEquals(call.timeout().timeoutNanos(), SECONDS.toNanos(1)); + } + private void givenResponseIsProduced(AsyncHttpClient client, Response response) { when(client.executeRequest(any(org.asynchttpclient.Request.class), any())).thenAnswer(invocation -> { AsyncCompletionHandler handler = invocation.getArgument(1); @@ -290,9 +367,11 @@ private void givenResponseIsProduced(AsyncHttpClient client, Response response) } private okhttp3.Response whenRequestIsMade(AsyncHttpClient client, Request request) throws IOException { - AsyncHttpClientCall call = AsyncHttpClientCall.builder().httpClient(client).request(request).build(); - - return call.execute(); + return AsyncHttpClientCall.builder() + .httpClientSupplier(() -> client) + .request(request) + .build() + .execute(); } private Request requestWithBody() { @@ -323,6 +402,13 @@ private Response responseWithBody(String contentType, String content) { return response; } + private Response responseWithNoBody() { + Response response = aResponse(); + when(response.hasResponseBody()).thenReturn(false); + when(response.getContentType()).thenReturn(null); + return response; + } + private void doThrow(String message) { throw new RuntimeException(message); } diff --git a/extras/simple/src/main/java/org/asynchttpclient/extras/simple/SimpleAsyncHttpClient.java b/extras/simple/src/main/java/org/asynchttpclient/extras/simple/SimpleAsyncHttpClient.java index b58658fb57..8d5bf18afe 100644 --- a/extras/simple/src/main/java/org/asynchttpclient/extras/simple/SimpleAsyncHttpClient.java +++ b/extras/simple/src/main/java/org/asynchttpclient/extras/simple/SimpleAsyncHttpClient.java @@ -380,7 +380,7 @@ public interface DerivedBuilder { DerivedBuilder setFormParams(Map> params); - DerivedBuilder setHeaders(Map> headers); + DerivedBuilder setHeaders(Map> headers); DerivedBuilder setHeaders(HttpHeaders headers); @@ -465,7 +465,7 @@ public Builder setHeaders(HttpHeaders headers) { return this; } - public Builder setHeaders(Map> headers) { + public Builder setHeaders(Map> headers) { requestBuilder.setHeaders(headers); return this; } diff --git a/extras/typesafeconfig/README.md b/extras/typesafeconfig/README.md index 3078cac2d5..dcc29dc269 100644 --- a/extras/typesafeconfig/README.md +++ b/extras/typesafeconfig/README.md @@ -8,7 +8,7 @@ Download [the latest JAR][2] or grab via [Maven][3]: ```xml org.asynchttpclient - async-http-client-extras-typesafeconfig + async-http-client-extras-typesafe-config latest.version ``` @@ -16,12 +16,12 @@ Download [the latest JAR][2] or grab via [Maven][3]: or [Gradle][3]: ```groovy -compile "org.asynchttpclient:async-http-client-extras-typesafeconfig:latest.version" +compile "org.asynchttpclient:async-http-client-extras-typesafe-config:latest.version" ``` [1]: https://github.com/lightbend/config - [2]: https://search.maven.org/remote_content?g=org.asynchttpclient&a=async-http-client-extras-typesafeconfig&v=LATEST - [3]: http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22org.asynchttpclient%22%20a%3A%22async-http-client-extras-typesafeconfig%22 + [2]: https://search.maven.org/remote_content?g=org.asynchttpclient&a=async-http-client-extras-typesafe-config&v=LATEST + [3]: http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22org.asynchttpclient%22%20a%3A%22async-http-client-extras-typesafe-config%22 [snap]: https://oss.sonatype.org/content/repositories/snapshots/ ## Example usage @@ -31,4 +31,4 @@ compile "org.asynchttpclient:async-http-client-extras-typesafeconfig:latest.vers com.typesafe.config.Config config = ... AsyncHttpClientTypesafeConfig ahcConfig = new AsyncHttpClientTypesafeConfig(config); AsyncHttpClient client = new DefaultAsyncHttpClient(ahcConfig); -``` \ No newline at end of file +``` diff --git a/extras/typesafeconfig/src/main/java/org/asynchttpclient/extras/typesafeconfig/AsyncHttpClientTypesafeConfig.java b/extras/typesafeconfig/src/main/java/org/asynchttpclient/extras/typesafeconfig/AsyncHttpClientTypesafeConfig.java index 8917052611..55c88ab251 100644 --- a/extras/typesafeconfig/src/main/java/org/asynchttpclient/extras/typesafeconfig/AsyncHttpClientTypesafeConfig.java +++ b/extras/typesafeconfig/src/main/java/org/asynchttpclient/extras/typesafeconfig/AsyncHttpClientTypesafeConfig.java @@ -69,6 +69,11 @@ public int getMaxConnectionsPerHost() { return getIntegerOpt(MAX_CONNECTIONS_PER_HOST_CONFIG).orElse(defaultMaxConnectionsPerHost()); } + @Override + public int getAcquireFreeChannelTimeout() { + return getIntegerOpt(ACQUIRE_FREE_CHANNEL_TIMEOUT).orElse(defaultAcquireFreeChannelTimeout()); + } + @Override public int getConnectTimeout() { return getIntegerOpt(CONNECTION_TIMEOUT_CONFIG).orElse(defaultConnectTimeout()); @@ -407,7 +412,7 @@ private Optional> getListOpt(String key) { private Optional getOpt(Function func, String key) { return config.hasPath(key) - ? Optional.ofNullable(func.apply(key)) - : Optional.empty(); + ? Optional.ofNullable(func.apply(key)) + : Optional.empty(); } } diff --git a/pom.xml b/pom.xml index 4b7d7b8bb7..48bb260312 100644 --- a/pom.xml +++ b/pom.xml @@ -80,7 +80,7 @@ com.gkatzioura.maven.cloud s3-storage-wagon - 1.8 + 2.1 install @@ -156,39 +156,39 @@ + + org.codehaus.mojo + flatten-maven-plugin + 1.0.0 + + true + + + + flatten + process-resources + + flatten + + + + flatten.clean + clean + + clean + + + + - org.codehaus.mojo - flatten-maven-plugin - 1.0.0 + org.apache.maven.plugins + maven-javadoc-plugin + 3.0.1 - true + none - - - flatten - process-resources - - flatten - - - - flatten.clean - clean - - clean - - - - - - - maven-javadoc-plugin - 2.10.4 - - - @@ -223,18 +223,16 @@ - - - dist-release - dist-release - ${distMgmtReleasesUrl} - - - dist-snapshots - dist-snapshots - ${distMgmtSnapshotsUrl} - - + + + dist-snapshots + ${distMgmtSnapshotsUrl} + + + dist-release + ${distMgmtReleasesUrl} + + netty-utils client @@ -320,6 +318,12 @@ rxjava ${rxjava2.version} + + org.apache.kerby + kerb-simplekdc + ${kerby.version} + test + @@ -420,34 +424,35 @@ org.hamcrest - java-hamcrest + hamcrest ${hamcrest.version} test - 2.5.3-SNAPSHOT + 2.8.1-SNAPSHOT http://oss.sonatype.org/content/repositories/snapshots http://oss.sonatype.org/content/repositories/releases true 1.8 1.8 - 4.1.27.Final + 4.1.33.Final 1.7.25 1.0.2 1.2.0 2.0.0 1.3.8 - 2.1.16 + 2.2.5 1.2.3 6.13.1 - 9.4.11.v20180605 - 9.0.10 + 9.4.14.v20181114 + 9.0.14 2.6 1.3.3 1.2.2 - 2.19.0 - 2.0.0.0 + 2.23.4 + 2.1 + 1.1.1