Ensure async data consumers can avoid NPE if they have been canceled / released from another thread at the same with concurrent data processing

Author: ok2c

httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/AbstractAsyncRequesterConsumer.java

@@ -113,19 +113,27 @@ public void completed(final E entity) {
     @Override
     public final void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
         final AsyncEntityConsumer<E> dataConsumer = dataConsumerRef.get();
-        dataConsumer.updateCapacity(capacityChannel);
+        if (dataConsumer != null) {
+            dataConsumer.updateCapacity(capacityChannel);
+        } else {
+            capacityChannel.update(Integer.MAX_VALUE);
+        }
     }
 
     @Override
     public final void consume(final ByteBuffer src) throws IOException {
         final AsyncEntityConsumer<E> dataConsumer = dataConsumerRef.get();
-        dataConsumer.consume(src);
+        if (dataConsumer != null) {
+            dataConsumer.consume(src);
+        }
     }
 
     @Override
     public final void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException {
         final AsyncEntityConsumer<E> dataConsumer = dataConsumerRef.get();
-        dataConsumer.streamEnd(trailers);
+        if (dataConsumer != null) {
+            dataConsumer.streamEnd(trailers);
+        }
     }
 
     @Override

httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/AbstractServerExchangeHandler.java

@@ -174,22 +174,27 @@ public void cancelled() {
     @Override
     public final void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
         final AsyncRequestConsumer<T> requestConsumer = requestConsumerRef.get();
-        Asserts.notNull(requestConsumer, "Data consumer");
-        requestConsumer.updateCapacity(capacityChannel);
+        if (requestConsumer != null) {
+            requestConsumer.updateCapacity(capacityChannel);
+        } else {
+            capacityChannel.update(Integer.MAX_VALUE);
+        }
     }
 
     @Override
     public final void consume(final ByteBuffer src) throws IOException {
         final AsyncRequestConsumer<T> requestConsumer = requestConsumerRef.get();
-        Asserts.notNull(requestConsumer, "Data consumer");
-        requestConsumer.consume(src);
+        if (requestConsumer != null) {
+            requestConsumer.consume(src);
+        }
     }
 
     @Override
     public final void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException {
         final AsyncRequestConsumer<T> requestConsumer = requestConsumerRef.get();
-        Asserts.notNull(requestConsumer, "Data consumer");
-        requestConsumer.streamEnd(trailers);
+        if (requestConsumer != null) {
+            requestConsumer.streamEnd(trailers);
+        }
     }
 
     @Override

httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/BasicRequestConsumer.java

@@ -100,19 +100,27 @@ public void completed(final T body) {
     @Override
     public void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
         final AsyncEntityConsumer<T> dataConsumer = dataConsumerRef.get();
-        dataConsumer.updateCapacity(capacityChannel);
+        if (dataConsumer != null) {
+            dataConsumer.updateCapacity(capacityChannel);
+        } else {
+            capacityChannel.update(Integer.MAX_VALUE);
+        }
     }
 
     @Override
     public void consume(final ByteBuffer src) throws IOException {
         final AsyncEntityConsumer<T> dataConsumer = dataConsumerRef.get();
-        dataConsumer.consume(src);
+        if (dataConsumer != null) {
+            dataConsumer.consume(src);
+        }
     }
 
     @Override
     public void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException {
         final AsyncEntityConsumer<T> dataConsumer = dataConsumerRef.get();
-        dataConsumer.streamEnd(trailers);
+        if (dataConsumer != null) {
+            dataConsumer.streamEnd(trailers);
+        }
     }
 
     @Override

httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/BasicResponseConsumer.java

@@ -104,19 +104,27 @@ public void informationResponse(final HttpResponse response, final HttpContext h
     @Override
     public void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
         final AsyncEntityConsumer<T> dataConsumer = dataConsumerRef.get();
-        dataConsumer.updateCapacity(capacityChannel);
+        if (dataConsumer != null) {
+            dataConsumer.updateCapacity(capacityChannel);
+        } else {
+            capacityChannel.update(Integer.MAX_VALUE);
+        }
     }
 
     @Override
     public void consume(final ByteBuffer src) throws IOException {
         final AsyncEntityConsumer<T> dataConsumer = dataConsumerRef.get();
-        dataConsumer.consume(src);
+        if (dataConsumer != null) {
+            dataConsumer.consume(src);
+        }
     }
 
     @Override
     public void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException {
         final AsyncEntityConsumer<T> dataConsumer = dataConsumerRef.get();
-        dataConsumer.streamEnd(trailers);
+        if (dataConsumer != null) {
+            dataConsumer.streamEnd(trailers);
+        }
     }
 
     @Override