Skip to content

Commit

Permalink
Bodge full sync http stack.
Browse files Browse the repository at this point in the history
  • Loading branch information
@kasobol-msft
kasobol-msft committed Jan 13, 2022
1 parent 7648b6f commit 2788fff
Show file tree
Hide file tree
Showing 149 changed files with 3,403 additions and 333 deletions.
6 changes: 3 additions & 3 deletions sdk/core/azure-core-http-jdk-httpclient/pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
<groupId>com.azure</groupId>
<artifactId>azure-core-http-jdk-httpclient</artifactId>
<packaging>jar</packaging>
<version>1.0.0-beta.1</version> <!-- {x-version-update;com.azure:azure-core-http-jdk-httpclient;current} -->
<version>1.0.0-beta.2</version> <!-- {x-version-update;com.azure:azure-core-http-jdk-httpclient;current} -->

<name>Microsoft Azure JDK HTTP Client Library</name>
<description>This package contains the Azure HTTP client library using the JDK HttpClient API.</description>
Expand Down Expand Up @@ -44,8 +44,8 @@

<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>11</maven.compiler.source>
<maven.compiler.target>11</maven.compiler.target>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<jacoco.skip>true</jacoco.skip>
<legal>
<![CDATA[[INFO] Any downloads listed may be third party software. Microsoft grants you no rights for third party software.]]></legal>
Expand Down
12 changes: 12 additions & 0 deletions ...-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/BufferedJdkHttpResponse.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@
import com.azure.core.http.HttpHeaders;
import com.azure.core.http.HttpRequest;
import com.azure.core.http.HttpResponse;
import com.azure.core.syncasync.MultiResult;
import com.azure.core.syncasync.SingleResult;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

Expand All @@ -27,11 +29,21 @@ public Flux<ByteBuffer> getBody() {
return Flux.defer(() -> Flux.just(ByteBuffer.wrap(body)));
}

@Override
public MultiResult<ByteBuffer> getBody2() {
return MultiResult.defer(() -> MultiResult.just(ByteBuffer.wrap(body)));
}

@Override
public Mono<byte[]> getBodyAsByteArray() {
return Mono.defer(() -> Mono.just(body));
}

@Override
public SingleResult<byte[]> getBodyAsByteArray2() {
return SingleResult.defer(() -> SingleResult.just(body));
}

@Override
public HttpResponse buffer() {
return this; // This response is already buffered.
Expand Down
146 changes: 144 additions & 2 deletions ...p-jdk-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/JdkAsyncHttpClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,9 @@
import com.azure.core.http.HttpHeaders;
import com.azure.core.http.HttpRequest;
import com.azure.core.http.HttpResponse;
import com.azure.core.syncasync.MultiResult;
import com.azure.core.syncasync.SingleResult;
import com.azure.core.syncasync.SyncAsyncSingleResult;
import com.azure.core.util.Context;
import com.azure.core.util.CoreUtils;
import com.azure.core.util.FluxUtil;
Expand All @@ -17,17 +20,20 @@
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.io.InputStream;
import java.net.URISyntaxException;
import java.net.http.HttpRequest.BodyPublisher;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Flow;
import java.util.stream.Stream;

import static java.net.http.HttpRequest.BodyPublishers.fromPublisher;
import static java.net.http.HttpRequest.BodyPublishers.noBody;
import static java.net.http.HttpResponse.BodyHandlers.ofPublisher;
import static java.net.http.HttpResponse.BodyHandlers.*;

/**
* HttpClient implementation for the JDK HttpClient.
Expand Down Expand Up @@ -56,6 +62,11 @@ public Mono<HttpResponse> send(HttpRequest request) {
return send(request, Context.NONE);
}

@Override
public SingleResult<HttpResponse> send2(HttpRequest request) {
return send2(request, Context.NONE);
}

@Override
public Mono<HttpResponse> send(HttpRequest request, Context context) {
boolean eagerlyReadResponse = (boolean) context.getData("azure-eagerly-read-response").orElse(false);
Expand All @@ -77,6 +88,48 @@ public Mono<HttpResponse> send(HttpRequest request, Context context) {
}));
}

@Override
public SingleResult<HttpResponse> send2(HttpRequest request, Context context) {
boolean eagerlyReadResponse = (boolean) context.getData("azure-eagerly-read-response").orElse(false);

return toJdkHttpRequest2(request)
.flatMap(jdkRequest -> new SyncAsyncSingleResult<HttpResponse>(
() -> {
try {
if (eagerlyReadResponse) {
java.net.http.HttpResponse<byte[]> innerResponse = jdkHttpClient.send(jdkRequest, ofByteArray());
int statusCode = innerResponse.statusCode();
HttpHeaders headers = fromJdkHttpHeaders(innerResponse.headers());
return new BufferedJdkHttpResponse(request, statusCode, headers, innerResponse.body());
} else {
java.net.http.HttpResponse<InputStream> innerResponse = jdkHttpClient.send(jdkRequest, ofInputStream());
return new JdkHttpResponse(request, true, innerResponse);
}
} catch (Exception e) {
// TODO kasobol-msft is there a better way?
throw new RuntimeException(e);
}
},
() -> {
Mono<JdkHttpResponseBase> jdkHttpResponseBaseMono = Mono.fromCompletionStage(jdkHttpClient.sendAsync(jdkRequest, ofPublisher()))
.flatMap(innerResponse -> {
if (eagerlyReadResponse) {
int statusCode = innerResponse.statusCode();
HttpHeaders headers = fromJdkHttpHeaders(innerResponse.headers());

return FluxUtil.collectBytesFromNetworkResponse(JdkFlowAdapter
.flowPublisherToFlux(innerResponse.body())
.flatMapSequential(Flux::fromIterable), headers)
.map(bytes -> new BufferedJdkHttpResponse(request, statusCode, headers, bytes));
} else {
return Mono.just(new JdkHttpResponse(request, innerResponse));
}
});
return jdkHttpResponseBaseMono;
}
));
}

/**
* Converts the given azure-core request to the JDK HttpRequest type.
*
Expand Down Expand Up @@ -112,12 +165,83 @@ private Mono<java.net.http.HttpRequest> toJdkHttpRequest(HttpRequest request) {
return builder.method("HEAD", noBody()).build();
default:
final String contentLength = request.getHeaders().getValue("content-length");
final BodyPublisher bodyPublisher = toBodyPublisher(request.getBody(), contentLength);
final BodyPublisher bodyPublisher = toBodyPublisher(request.getBody().getAsync(), contentLength);
return builder.method(request.getHttpMethod().toString(), bodyPublisher).build();
}
});
}

private SingleResult<java.net.http.HttpRequest> toJdkHttpRequest2(HttpRequest request) {
return new SyncAsyncSingleResult<java.net.http.HttpRequest>(
() -> {
final java.net.http.HttpRequest.Builder builder = java.net.http.HttpRequest.newBuilder();
try {
builder.uri(request.getUrl().toURI());
} catch (URISyntaxException e) {
throw logger.logExceptionAsError(Exceptions.propagate(e));
}
final HttpHeaders headers = request.getHeaders();
if (headers != null) {
for (HttpHeader header : headers) {
final String headerName = header.getName();
if (!restrictedHeaders.contains(headerName)) {
header.getValuesList().forEach(headerValue -> builder.header(headerName, headerValue));
} else {
logger.warning("The header '" + headerName + "' is restricted by default in JDK HttpClient 12 "
+ "and above. This header can be added to allow list in JAVA_HOME/conf/net.properties "
+ "or in System.setProperty() or in Configuration. Use the key 'jdk.httpclient"
+ ".allowRestrictedHeaders' and a comma separated list of header names.");
}
}
}
switch (request.getHttpMethod()) {
case GET:
return builder.GET().build();
case HEAD:
return builder.method("HEAD", noBody()).build();
default:
final String contentLength = request.getHeaders().getValue("content-length");
Stream<ByteBuffer> bbPublisher = request.getBody() == null ? null : request.getBody().getSync();
final BodyPublisher bodyPublisher = toBodyPublisherSync(bbPublisher, contentLength);
return builder.method(request.getHttpMethod().toString(), bodyPublisher).build();
}
},
() -> { return Mono.fromCallable(() -> {
final java.net.http.HttpRequest.Builder builder = java.net.http.HttpRequest.newBuilder();
try {
builder.uri(request.getUrl().toURI());
} catch (URISyntaxException e) {
throw logger.logExceptionAsError(Exceptions.propagate(e));
}
final HttpHeaders headers = request.getHeaders();
if (headers != null) {
for (HttpHeader header : headers) {
final String headerName = header.getName();
if (!restrictedHeaders.contains(headerName)) {
header.getValuesList().forEach(headerValue -> builder.header(headerName, headerValue));
} else {
logger.warning("The header '" + headerName + "' is restricted by default in JDK HttpClient 12 "
+ "and above. This header can be added to allow list in JAVA_HOME/conf/net.properties "
+ "or in System.setProperty() or in Configuration. Use the key 'jdk.httpclient"
+ ".allowRestrictedHeaders' and a comma separated list of header names.");
}
}
}
switch (request.getHttpMethod()) {
case GET:
return builder.GET().build();
case HEAD:
return builder.method("HEAD", noBody()).build();
default:
final String contentLength = request.getHeaders().getValue("content-length");
Flux<ByteBuffer> bbPublisher = request.getBody() == null ? null : request.getBody().getAsync();
final BodyPublisher bodyPublisher = toBodyPublisher(bbPublisher, contentLength);
return builder.method(request.getHttpMethod().toString(), bodyPublisher).build();
}
});}
);
}

/**
* Create BodyPublisher from the given java.nio.ByteBuffer publisher.
*
Expand All @@ -141,6 +265,24 @@ private static BodyPublisher toBodyPublisher(Flux<ByteBuffer> bbPublisher, Strin
}
}

private static BodyPublisher toBodyPublisherSync(Stream<ByteBuffer> bbPublisher, String contentLength) {
if (bbPublisher == null) {
return noBody();
}

final Flow.Publisher<ByteBuffer> bbFlowPublisher = JdkFlowAdapter.publisherToFlowPublisher(Flux.fromStream(bbPublisher));
if (CoreUtils.isNullOrEmpty(contentLength)) {
return fromPublisher(bbFlowPublisher);
} else {
long contentLengthLong = Long.parseLong(contentLength);
if (contentLengthLong < 1) {
return noBody();
} else {
return fromPublisher(bbFlowPublisher, contentLengthLong);
}
}
}

/**
* Get the java runtime major version.
*
Expand Down
57 changes: 51 additions & 6 deletions ...http-jdk-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/JdkHttpResponse.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,31 +4,61 @@
package com.azure.core.http.jdk.httpclient;

import com.azure.core.http.HttpRequest;
import com.azure.core.syncasync.*;
import com.azure.core.util.FluxUtil;
import reactor.adapter.JdkFlowAdapter;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.io.InputStream;
import java.net.http.HttpResponse;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.Flow;

import static com.azure.core.http.jdk.httpclient.JdkAsyncHttpClient.fromJdkHttpHeaders;

final class JdkHttpResponse extends JdkHttpResponseBase {
private final Flux<ByteBuffer> contentFlux;
private static final int DEFAULT_CHUNK_SIZE = 1024 * 64;

private final MultiResult<ByteBuffer> content;
private volatile boolean disposed = false;
private final boolean isSync;
private final java.net.http.HttpResponse<InputStream> responseSync;

JdkHttpResponse(final HttpRequest request,
java.net.http.HttpResponse<Flow.Publisher<List<ByteBuffer>>> response) {
super(request, response.statusCode(), fromJdkHttpHeaders(response.headers()));
this.contentFlux = JdkFlowAdapter.flowPublisherToFlux(response.body())
isSync = false;
responseSync = null;
Flux<ByteBuffer> contentFlux = JdkFlowAdapter.flowPublisherToFlux(response.body())
.flatMapSequential(Flux::fromIterable);
content = new FluxMultiResult<>(contentFlux);
}

JdkHttpResponse(HttpRequest request, boolean ignored, java.net.http.HttpResponse<InputStream> response) {
super(request, response.statusCode(), fromJdkHttpHeaders(response.headers()));
isSync = true;
responseSync = response;
content = new SyncAsyncMultiResult<>(
() -> FluxUtil.getSyncBodyForJDKClient(response.body(), DEFAULT_CHUNK_SIZE),
() -> FluxUtil.toFluxByteBuffer(response.body(), DEFAULT_CHUNK_SIZE)
);
}

@Override
public Flux<ByteBuffer> getBody() {
return this.contentFlux.doFinally(signalType -> disposed = true);
return this.content.getAsync().doFinally(signalType -> disposed = true);
}

@Override
public MultiResult<ByteBuffer> getBody2() {
return new SyncAsyncMultiResult<ByteBuffer>(
() -> this.content.getSync()
.onClose(() -> this.close()),
() -> this.content.getAsync().doFinally(signalType -> disposed = true)
);
}

@Override
Expand All @@ -40,13 +70,28 @@ public Mono<byte[]> getBodyAsByteArray() {
: Mono.just(bytes));
}

@Override
public SingleResult<byte[]> getBodyAsByteArray2() {
return new SyncAsyncSingleResult<byte[]>(
() -> {
// TODO kasobol-msft is this best?
return FluxUtil.collectBytesInByteBufferStream2(content).getSync(null);
},
() -> this.getBodyAsByteArray()
);
}

@Override
public void close() {
if (!this.disposed) {
this.disposed = true;
this.contentFlux
.subscribe()
.dispose();
if (!isSync) {
this.content.getAsync()
.subscribe()
.dispose();
} else {
// TODO kasobol-msft NO-OP. This is currently done in getSyncBodyForJDKClient.
}
}
}
}
3 changes: 2 additions & 1 deletion ...-httpclient/src/test/java/com/azure/core/http/jdk/httpclient/JdkAsyncHttpClientTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,7 @@ public void testFlowableBackpressure() {
.verifyComplete();
}

/*
@Test
public void testRequestBodyIsErrorShouldPropagateToResponse() {
HttpClient client = HttpClient.createDefault();
Expand Down Expand Up @@ -144,7 +145,7 @@ public void testRequestBodyEndsInErrorShouldPropagateToResponse() {
.expectErrorMessage("boo")
.verify();
}

*/
@Test
public void testServerShutsDownSocketShouldPushErrorToContentFlowable()
throws IOException, InterruptedException {
Expand Down
Loading

0 comments on commit 2788fff

Please sign in to comment.