Support SSE in http-netty4 and add more testing (Azure#44996)

Support SSE in http-netty4 and add more testing

Author: alzimmermsft

sdk/clientcore/annotation-processor-test/src/main/java/io/clientcore/annotation/processor/test/SimpleXmlSerializableServiceImpl.java renamed to sdk/clientcore/annotation-processor-test/src/main/java/io/clientcore/annotation/processor/test/test/SimpleXmlSerializableServiceImpl.java

@@ -38,17 +38,13 @@
   <properties>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <legal><![CDATA[[INFO] Any downloads listed may be third party software.  Microsoft grants you no rights for third party software.]]></legal>
-    <jacoco.min.linecoverage>0.75</jacoco.min.linecoverage>
-    <jacoco.min.branchcoverage>0.65</jacoco.min.branchcoverage>
+    <jacoco.min.linecoverage>0.85</jacoco.min.linecoverage>
+    <jacoco.min.branchcoverage>0.75</jacoco.min.branchcoverage>
     <javaModulesSurefireArgLine>
-      --add-exports io.clientcore.core/io.clientcore.core.implementation.http=ALL-UNNAMED
-      --add-exports io.clientcore.core/io.clientcore.core.implementation.http.serializer=ALL-UNNAMED
-      --add-exports io.clientcore.core/io.clientcore.core.implementation.util=ALL-UNNAMED
+      --add-opens io.clientcore.http.netty4/io.clientcore.http.netty4=ALL-UNNAMED
+      --add-opens io.clientcore.http.netty4/io.clientcore.http.netty4.implementation=ALL-UNNAMED
 
-      --add-opens io.clientcore.http.netty/io.clientcore.http.netty=ALL-UNNAMED
-      --add-opens io.clientcore.http.netty/io.clientcore.http.netty4.implementation=ALL-UNNAMED
-
-      --add-reads io.clientcore.http.netty=java.logging
+      --add-reads io.clientcore.http.netty4=java.logging
     </javaModulesSurefireArgLine>
 
     <!-- Enables fail on deprecated API usage. -->

sdk/clientcore/annotation-processor-test/src/main/java/io/clientcore/annotation/processor/test/implementation/TestInterfaceClientServiceImpl.java renamed to sdk/clientcore/annotation-processor-test/src/main/java/io/clientcore/annotation/processor/test/test/implementation/TestInterfaceClientServiceImpl.java

@@ -7,13 +7,16 @@
 import io.clientcore.core.http.models.HttpHeaderName;
 import io.clientcore.core.http.models.HttpRequest;
 import io.clientcore.core.http.models.Response;
+import io.clientcore.core.http.models.ServerSentEventListener;
 import io.clientcore.core.instrumentation.logging.ClientLogger;
 import io.clientcore.core.models.CoreException;
+import io.clientcore.core.models.ServerSentResult;
 import io.clientcore.core.models.binarydata.BinaryData;
 import io.clientcore.core.models.binarydata.FileBinaryData;
 import io.clientcore.core.models.binarydata.InputStreamBinaryData;
 import io.clientcore.core.utils.AuthenticateChallenge;
 import io.clientcore.core.utils.ProgressReporter;
+import io.clientcore.core.utils.ServerSentEventUtils;
 import io.clientcore.http.netty4.implementation.ChannelInitializationProxyHandler;
 import io.clientcore.http.netty4.implementation.Netty4ProgressAndTimeoutHandler;
 import io.clientcore.http.netty4.implementation.Netty4ResponseHandler;
@@ -31,6 +34,7 @@
 import io.netty.handler.codec.http.DefaultHttpRequest;
 import io.netty.handler.codec.http.HttpMethod;
 import io.netty.handler.codec.http.HttpVersion;
+import io.netty.handler.proxy.ProxyHandler;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.handler.stream.ChunkedInput;
@@ -47,8 +51,11 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static io.clientcore.core.utils.ServerSentEventUtils.attemptRetry;
+import static io.clientcore.core.utils.ServerSentEventUtils.processTextEventStream;
 import static io.clientcore.http.netty4.implementation.Netty4Utility.PROGRESS_AND_TIMEOUT_HANDLER_NAME;
 import static io.clientcore.http.netty4.implementation.Netty4Utility.createCodec;
+import static io.clientcore.http.netty4.implementation.Netty4Utility.setOrSuppressError;
 import static io.netty.handler.codec.http.DefaultHttpHeadersFactory.trailersFactory;
 
 /**
@@ -57,6 +64,12 @@
 class NettyHttpClient implements HttpClient {
     private static final ClientLogger LOGGER = new ClientLogger(NettyHttpClient.class);
 
+    /**
+     * Error message for when no {@link ServerSentEventListener} is attached to the {@link HttpRequest}.
+     */
+    private static final String NO_LISTENER_ERROR_MESSAGE
+        = "No ServerSentEventListener attached to HttpRequest to handle the text/event-stream response";
+
     private final Bootstrap bootstrap;
     private final SslContext sslContext;
     private final ChannelInitializationProxyHandler channelInitializationProxyHandler;
@@ -93,14 +106,26 @@ public Response<BinaryData> send(HttpRequest request) {
         boolean addProgressAndTimeoutHandler
             = progressReporter != null || writeTimeoutMillis > 0 || responseTimeoutMillis > 0 || readTimeoutMillis > 0;
 
+        AtomicReference<Response<BinaryData>> responseReference = new AtomicReference<>();
+        AtomicReference<Throwable> errorReference = new AtomicReference<>();
+        CountDownLatch latch = new CountDownLatch(1);
+
         // Configure an immutable ChannelInitializer in the builder with everything that can be added on a non-per
         // request basis.
         bootstrap.handler(new ChannelInitializer<Channel>() {
             @Override
             protected void initChannel(Channel ch) throws SSLException {
                 // Test whether proxying should be applied to this Channel. If so, add it.
-                if (channelInitializationProxyHandler.test(ch.remoteAddress())) {
-                    ch.pipeline().addFirst(channelInitializationProxyHandler.createProxy(proxyChallenges));
+                boolean hasProxy = channelInitializationProxyHandler.test(ch.remoteAddress());
+                if (hasProxy) {
+                    ProxyHandler proxyHandler = channelInitializationProxyHandler.createProxy(proxyChallenges);
+                    proxyHandler.connectFuture().addListener(future -> {
+                        if (!future.isSuccess()) {
+                            setOrSuppressError(errorReference, future.cause());
+                        }
+                    });
+
+                    ch.pipeline().addFirst(proxyHandler);
                 }
 
                 // Add SSL handling if the request is HTTPS.
@@ -118,35 +143,56 @@ protected void initChannel(Channel ch) throws SSLException {
             }
         });
 
-        AtomicReference<Response<BinaryData>> responseReference = new AtomicReference<>();
-        AtomicReference<Throwable> errorReference = new AtomicReference<>();
-        CountDownLatch latch = new CountDownLatch(1);
-
         try {
-            Channel channel = bootstrap.connect(host, port).sync().channel();
-
-            // Only add CoreProgressAndTimeoutHandler if it will do anything, ex it is reporting progress or is
-            // applying timeouts.
-            // This is done to keep the ChannelPipeline shorter, therefore more performant, if this would
-            // effectively be a no-op.
-            if (addProgressAndTimeoutHandler) {
-                channel.pipeline()
-                    .addLast(PROGRESS_AND_TIMEOUT_HANDLER_NAME, new Netty4ProgressAndTimeoutHandler(progressReporter,
-                        writeTimeoutMillis, responseTimeoutMillis, readTimeoutMillis));
-            }
+            bootstrap.connect(host, port).addListener((ChannelFutureListener) connectListener -> {
+                if (!connectListener.isSuccess()) {
+                    LOGGER.atError().setThrowable(connectListener.cause()).log("Failed to send request");
+                    errorReference.set(connectListener.cause());
+                    connectListener.channel().close();
+                    latch.countDown();
+                    return;
+                }
 
-            Netty4ResponseHandler responseHandler = new Netty4ResponseHandler(request, responseReference, latch);
-            channel.pipeline().addLast(responseHandler);
+                Channel channel = connectListener.channel();
+                channel.closeFuture().addListener(closeListener -> {
+                    if (!closeListener.isSuccess()) {
+                        LOGGER.atError().setThrowable(closeListener.cause()).log("Channel closed with error");
+                        setOrSuppressError(errorReference, closeListener.cause());
+                    }
+                });
+
+                // Only add CoreProgressAndTimeoutHandler if it will do anything, ex it is reporting progress or is
+                // applying timeouts.
+                // This is done to keep the ChannelPipeline shorter, therefore more performant, if this would
+                // effectively be a no-op.
+                if (addProgressAndTimeoutHandler) {
+                    channel.pipeline()
+                        .addLast(PROGRESS_AND_TIMEOUT_HANDLER_NAME, new Netty4ProgressAndTimeoutHandler(
+                            progressReporter, writeTimeoutMillis, responseTimeoutMillis, readTimeoutMillis));
+                }
 
-            sendRequest(request, channel, addProgressAndTimeoutHandler).addListener((ChannelFutureListener) future -> {
-                if (!future.isSuccess()) {
-                    LOGGER.atError().setThrowable(future.cause()).log("Failed to send request");
-                    errorReference.set(future.cause());
-                    future.channel().close();
+                Netty4ResponseHandler responseHandler
+                    = new Netty4ResponseHandler(request, responseReference, errorReference, latch);
+                channel.pipeline().addLast(responseHandler);
+
+                Throwable earlyError = errorReference.get();
+                if (earlyError != null) {
+                    // If an error occurred between the connect and the request being sent, don't proceed with sending
+                    // the request.
                     latch.countDown();
-                } else {
-                    future.channel().read();
+                    return;
                 }
+
+                sendRequest(request, channel, addProgressAndTimeoutHandler, errorReference)
+                    .addListener((ChannelFutureListener) sendListener -> {
+                        if (!sendListener.isSuccess()) {
+                            setOrSuppressError(errorReference, sendListener.cause());
+                            sendListener.channel().close();
+                            latch.countDown();
+                        } else {
+                            sendListener.channel().read();
+                        }
+                    });
             });
 
             latch.await();
@@ -159,10 +205,42 @@ protected void initChannel(Channel ch) throws SSLException {
         if (response == null) {
             throw CoreException.from(errorReference.get());
         }
+
+        if (response.getValue() != BinaryData.empty()
+            && ServerSentEventUtils
+                .isTextEventStreamContentType(response.getHeaders().getValue(HttpHeaderName.CONTENT_TYPE))) {
+            ServerSentEventListener listener = request.getServerSentEventListener();
+
+            if (listener != null) {
+                try {
+                    ServerSentResult serverSentResult
+                        = processTextEventStream(response.getValue().toStream(), listener);
+
+                    if (serverSentResult.getException() != null) {
+                        // If an exception occurred while processing the text event stream, emit listener onError.
+                        listener.onError(serverSentResult.getException());
+                    }
+
+                    // If an error occurred or we want to reconnect
+                    if (!Thread.currentThread().isInterrupted() && attemptRetry(serverSentResult, request)) {
+                        return this.send(request);
+                    }
+
+                    response = new Response<>(response.getRequest(), response.getStatusCode(), response.getHeaders(),
+                        createBodyFromServerSentResult(serverSentResult));
+                } catch (IOException ex) {
+                    throw LOGGER.logThrowableAsError(CoreException.from(ex));
+                }
+            } else {
+                throw LOGGER.logThrowableAsError(new IllegalStateException(NO_LISTENER_ERROR_MESSAGE));
+            }
+        }
+
         return response;
     }
 
-    private ChannelFuture sendRequest(HttpRequest request, Channel channel, boolean progressAndTimeoutHandlerAdded) {
+    private ChannelFuture sendRequest(HttpRequest request, Channel channel, boolean progressAndTimeoutHandlerAdded,
+        AtomicReference<Throwable> errorReference) {
         HttpMethod nettyMethod = HttpMethod.valueOf(request.getHttpMethod().toString());
         String uri = request.getUri().toString();
         WrappedHttpHeaders wrappedHttpHeaders = new WrappedHttpHeaders(request.getHeaders());
@@ -179,35 +257,41 @@ private ChannelFuture sendRequest(HttpRequest request, Channel channel, boolean
                     new ChunkedNioFile(FileChannel.open(fileBinaryData.getFile(), StandardOpenOption.READ),
                         fileBinaryData.getPosition(), fileBinaryData.getLength(), 8192),
                     new DefaultHttpRequest(HttpVersion.HTTP_1_1, nettyMethod, uri, wrappedHttpHeaders),
-                    progressAndTimeoutHandlerAdded);
+                    progressAndTimeoutHandlerAdded, errorReference);
             } catch (IOException ex) {
                 return channel.newFailedFuture(ex);
             }
         } else if (requestBody instanceof InputStreamBinaryData) {
             return sendChunked(channel, new ChunkedStream(requestBody.toStream()),
                 new DefaultHttpRequest(HttpVersion.HTTP_1_1, nettyMethod, uri, wrappedHttpHeaders),
-                progressAndTimeoutHandlerAdded);
-        }
+                progressAndTimeoutHandlerAdded, errorReference);
+        } else {
+            ByteBuf body = Unpooled.EMPTY_BUFFER;
+            if (requestBody != null && requestBody != BinaryData.empty()) {
+                // Longer term, see if there is a way to have BinaryData act as the ByteBuf body to further eliminate
+                // copying of byte[]s.
+                body = Unpooled.wrappedBuffer(requestBody.toBytes());
+            }
+            if (body.readableBytes() > 0) {
+                // TODO (alzimmer): Should we be setting Content-Length here again? Shouldn't this be handled externally
+                //  by the creator of the HttpRequest?
+                wrappedHttpHeaders.getCoreHeaders()
+                    .set(HttpHeaderName.CONTENT_LENGTH, String.valueOf(body.readableBytes()));
+            }
 
-        ByteBuf body = Unpooled.EMPTY_BUFFER;
-        if (requestBody != null && requestBody != BinaryData.empty()) {
-            // Longer term, see if there is a way to have BinaryData act as the ByteBuf body to further eliminate
-            // copying of byte[]s.
-            body = Unpooled.wrappedBuffer(requestBody.toBytes());
-        }
-        if (body.readableBytes() > 0) {
-            // TODO (alzimmer): Should we be setting Content-Length here again? Shouldn't this be handled externally
-            //  by the creator of the HttpRequest?
-            wrappedHttpHeaders.getCoreHeaders()
-                .set(HttpHeaderName.CONTENT_LENGTH, String.valueOf(body.readableBytes()));
-        }
+            Throwable error = errorReference.get();
+            if (error != null) {
+                return channel.newFailedFuture(error);
+            }
 
-        return channel.writeAndFlush(new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, nettyMethod, uri, body,
-            wrappedHttpHeaders, trailersFactory().newHeaders()));
+            return channel.writeAndFlush(new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, nettyMethod, uri, body,
+                wrappedHttpHeaders, trailersFactory().newHeaders()));
+        }
     }
 
     private <T> ChannelFuture sendChunked(Channel channel, ChunkedInput<T> chunkedInput,
-        io.netty.handler.codec.http.HttpRequest initialLineAndHeaders, boolean progressAndTimeoutHandlerAdded) {
+        io.netty.handler.codec.http.HttpRequest initialLineAndHeaders, boolean progressAndTimeoutHandlerAdded,
+        AtomicReference<Throwable> errorReference) {
         if (channel.pipeline().get(ChunkedWriteHandler.class) == null) {
             // Add the ChunkedWriteHandler which will handle sending the chunkedInput.
             ChunkedWriteHandler chunkedWriteHandler = new ChunkedWriteHandler();
@@ -218,6 +302,11 @@ private <T> ChannelFuture sendChunked(Channel channel, ChunkedInput<T> chunkedIn
             }
         }
 
+        Throwable error = errorReference.get();
+        if (error != null) {
+            return channel.newFailedFuture(error);
+        }
+
         channel.write(initialLineAndHeaders);
         return channel.writeAndFlush(chunkedInput);
     }
@@ -228,4 +317,10 @@ public void close() {
             group.shutdownGracefully();
         }
     }
+
+    private static BinaryData createBodyFromServerSentResult(ServerSentResult serverSentResult) {
+        return (serverSentResult != null && serverSentResult.getData() != null)
+            ? BinaryData.fromString(String.join("\n", serverSentResult.getData()))
+            : BinaryData.empty();
+    }
 }

sdk/clientcore/http-netty4/pom.xml

@@ -10,6 +10,7 @@
 import io.clientcore.http.netty4.NettyHttpClientBuilder;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelPipeline;
+import io.netty.handler.proxy.HttpProxyHandler;
 import io.netty.handler.proxy.ProxyHandler;
 import io.netty.handler.proxy.Socks4ProxyHandler;
 import io.netty.handler.proxy.Socks5ProxyHandler;
@@ -83,8 +84,10 @@ public ProxyHandler createProxy(AtomicReference<List<AuthenticateChallenge>> pro
         } else if (proxyOptions.getType() == ProxyOptions.Type.SOCKS5) {
             return new Socks5ProxyHandler(proxyOptions.getAddress(), proxyOptions.getUsername(),
                 proxyOptions.getPassword());
-        } else {
+        } else if (proxyOptions.getUsername() != null && proxyOptions.getPassword() != null) {
             return new Netty4HttpProxyHandler(proxyOptions, proxyChallenges);
+        } else {
+            return new HttpProxyHandler(proxyOptions.getAddress());
         }
     }
 }