stream cacnel

Author: TingDaoK

crt/aws-c-auth

@@ -97,7 +97,23 @@ public int getResponseStatusCode() {
     }
 
     /**
-     * Cancels the stream.
+     * Cancels the stream with the default error code (AWS_ERROR_HTTP_STREAM_CANCELLED).
+     * <p>
+     * For HTTP/1.1 streams, this is equivalent to closing the connection.
+     * For HTTP/2 streams, this sends a RST_STREAM frame with AWS_HTTP2_ERR_CANCEL.
+     * <p>
+     * The stream will complete with AWS_ERROR_HTTP_STREAM_CANCELLED, unless the stream is
+     * already completing for other reasons, or the stream is not activated,
+     * in which case this call will have no effect.
+     */
+    public void cancel() {
+        if (!isNull()) {
+            httpStreamBaseCancelDefaultError(getNativeHandle());
+        }
+    }
+
+    /**
+     * Cancels the stream with a specific error code.
      * <p>
      * For HTTP/1.1 streams, this is equivalent to closing the connection.
      * For HTTP/2 streams, this sends a RST_STREAM frame with AWS_HTTP2_ERR_CANCEL.
@@ -107,7 +123,6 @@ public int getResponseStatusCode() {
      * in which case this call will have no effect.
      *
      * @param errorCode The CRT error code to use when completing the stream.
-     *                  Use CRT.awsErrorCode() to convert AWS error codes.
      */
     public void cancel(int errorCode) {
         if (!isNull()) {
@@ -127,5 +142,7 @@ public void cancel(int errorCode) {
 
     private static native int httpStreamBaseGetResponseStatusCode(long http_stream);
 
+    private static native void httpStreamBaseCancelDefaultError(long http_stream);
+
     private static native void httpStreamBaseCancel(long http_stream, int error_code);
 }

crt/aws-c-event-stream

@@ -706,11 +706,10 @@ JNIEXPORT void JNICALL Java_software_amazon_awssdk_crt_http_HttpStreamBase_httpS
     aws_http_stream_update_window(stream, window_update);
 }
 
-JNIEXPORT void JNICALL Java_software_amazon_awssdk_crt_http_HttpStreamBase_httpStreamBaseCancel(
+JNIEXPORT void JNICALL Java_software_amazon_awssdk_crt_http_HttpStreamBase_httpStreamBaseCancelDefaultError(
     JNIEnv *env,
     jclass jni_class,
-    jlong jni_binding,
-    jint error_code) {
+    jlong jni_binding) {
 
     (void)jni_class;
     aws_cache_jni_ids(env);
@@ -723,8 +722,8 @@ JNIEXPORT void JNICALL Java_software_amazon_awssdk_crt_http_HttpStreamBase_httpS
         return;
     }
 
-    AWS_LOGF_TRACE(AWS_LS_HTTP_STREAM, "Cancelling Stream. stream: %p, error_code: %d", (void *)stream, error_code);
-    aws_http_stream_cancel(stream, error_code);
+    AWS_LOGF_TRACE(AWS_LS_HTTP_STREAM, "Cancelling Stream with default error. stream: %p", (void *)stream);
+    aws_http_stream_cancel_default_error(stream);
 }
 
 JNIEXPORT void JNICALL Java_software_amazon_awssdk_crt_http_Http2Stream_http2StreamResetStream(

crt/aws-c-http

@@ -341,44 +341,4 @@ public void onResponseComplete(HttpStreamBase stream, int errorCode) {
         CrtResource.waitForNoResources();
     }
 
-    @Test
-    public void testHttp2StreamCancel() throws Exception {
-        skipIfAndroid();
-        skipIfLocalhostUnavailable();
-        URI uri = new URI(String.format("https://localhost:%d/echo", LOCAL_HTTPS_PORT));
-        try (Http2StreamManager streamManager = createStreamManager(uri, 100)) {
-            long bodyLength = 250000L;
-
-            Http2Request request = createHttp2Request("GET", uri, 0);
-            request.addHeader(new HttpHeader("x-repeat-data", String.valueOf(bodyLength)));
-            /* Get a slow response to make sure we cancel before finishes */
-            request.addHeader(new HttpHeader("x-slow-response", "true"));
-
-            final CompletableFuture<Void> requestCompleteFuture = new CompletableFuture<Void>();
-            final int expectedErrorCode = 0x0832; // Some Random error code
-            CompletableFuture<Http2Stream> acquireCompleteFuture = streamManager.acquireStream(request,
-                    new HttpStreamBaseResponseHandler() {
-                        @Override
-                        public void onResponseHeaders(HttpStreamBase stream, int responseStatusCode, int blockType,
-                                HttpHeader[] nextHeaders) {
-                            // Cancel the HTTP/2 stream immediately upon receiving headers with specific
-                            // error code
-                            stream.cancel(expectedErrorCode);
-                        }
-
-                        @Override
-                        public void onResponseComplete(HttpStreamBase stream, int errorCode) {
-
-                            Assert.assertTrue(errorCode == expectedErrorCode);
-                            stream.close();
-                            requestCompleteFuture.complete(null);
-                        }
-                    });
-
-            acquireCompleteFuture.get(3, TimeUnit.SECONDS);
-            requestCompleteFuture.get(10, TimeUnit.SECONDS);
-        }
-        CrtResource.logNativeResources();
-        CrtResource.waitForNoResources();
-    }
 }

crt/aws-c-mqtt

@@ -293,4 +293,48 @@ public void onResponseComplete(HttpStreamBase stream, int errorCode) {
         CrtResource.logNativeResources();
         CrtResource.waitForNoResources();
     }
+
+    @Test
+    public void testHttp2StreamCancel() throws Exception {
+        skipIfAndroid();
+        skipIfLocalhostUnavailable();
+
+        URI uri = new URI(endpoint);
+        String large_file_path = "/crt-canary-obj.txt";
+        int maxConcurrentStreams = 20;
+        try (Http2StreamManager streamManager = createStreamManager(uri, 100, maxConcurrentStreams)) {
+
+            Http2Request request = createHttp2Request("GET", endpoint, large_file_path, EMPTY_BODY);
+
+            final CompletableFuture<Integer> requestCompleteFuture = new CompletableFuture<>();
+
+            streamManager.acquireStream(request, new HttpStreamBaseResponseHandler() {
+                @Override
+                public void onResponseHeaders(HttpStreamBase stream, int responseStatusCode, int blockType,
+                        HttpHeader[] nextHeaders) {
+                    // Cancel the HTTP/2 stream using the default error code (AWS_ERROR_HTTP_STREAM_CANCELLED)
+                    stream.cancel();
+                }
+
+                @Override
+                public void onResponseComplete(HttpStreamBase stream, int errorCode) {
+                    requestCompleteFuture.complete(errorCode);
+                    stream.close();
+                }
+            }).whenComplete((stream, throwable) -> {
+                if (throwable != null) {
+                    requestCompleteFuture.completeExceptionally(throwable);
+                }
+            });
+
+
+            int actualErrorCode = requestCompleteFuture.get(60, TimeUnit.SECONDS);
+
+            // The HTTP/2 stream should complete with AWS_ERROR_HTTP_STREAM_CANCELLED
+            Assert.assertEquals("Error code should be AWS_ERROR_HTTP_STREAM_CANCELLED",
+                    "AWS_ERROR_HTTP_STREAM_CANCELLED", CRT.awsErrorName(actualErrorCode));
+        }
+        CrtResource.logNativeResources();
+        CrtResource.waitForNoResources();
+    }
 }