src/java.httpclient/share/classes/java/net/http/RawChannel.java

*** 26,59 ****
  import java.io.IOException;
  import java.nio.ByteBuffer;
  import java.nio.channels.ByteChannel;
  import java.nio.channels.GatheringByteChannel;
  import java.nio.channels.SelectableChannel;
  
! /**
!  * Used to implement WebSocket. Each RawChannel corresponds to
!  * a TCP connection (SocketChannel) but is connected to a Selector
!  * and an ExecutorService for invoking the send and receive callbacks
!  * Also includes SSL processing.
!  */
! class RawChannel implements ByteChannel, GatheringByteChannel {
  
      private final HttpClientImpl client;
      private final HttpConnection connection;
!     private boolean closed;
  
      private interface RawEvent {
  
          /** must return the selector interest op flags OR'd. */
          int interestOps();
  
          /** called when event occurs. */
          void handle();
      }
  
-     interface BlockingEvent extends RawEvent { }
- 
      interface NonBlockingEvent extends RawEvent { }
  
      RawChannel(HttpClientImpl client, HttpConnection connection) {
          this.client = client;
          this.connection = connection;
--- 26,57 ----
  import java.io.IOException;
  import java.nio.ByteBuffer;
  import java.nio.channels.ByteChannel;
  import java.nio.channels.GatheringByteChannel;
  import java.nio.channels.SelectableChannel;
+ import java.nio.channels.SelectionKey;
  
! //
! // Used to implement WebSocket. Each RawChannel corresponds to a TCP connection
! // (SocketChannel) but is connected to a Selector and an ExecutorService for
! // invoking the send and receive callbacks. Also includes SSL processing.
! //
! final class RawChannel implements ByteChannel, GatheringByteChannel {
  
      private final HttpClientImpl client;
      private final HttpConnection connection;
!     private volatile boolean closed;
  
      private interface RawEvent {
  
          /** must return the selector interest op flags OR'd. */
          int interestOps();
  
          /** called when event occurs. */
          void handle();
      }
  
      interface NonBlockingEvent extends RawEvent { }
  
      RawChannel(HttpClientImpl client, HttpConnection connection) {
          this.client = client;
          this.connection = connection;
*** 62,123 ****
      private class RawAsyncEvent extends AsyncEvent {
  
          private final RawEvent re;
  
          RawAsyncEvent(RawEvent re) {
              this.re = re;
          }
  
          public SelectableChannel channel() {
              return connection.channel();
          }
  
          // must return the selector interest op flags OR'd
          public int interestOps() {
              return re.interestOps();
          }
  
          // called when event occurs
          public void handle() {
              re.handle();
          }
  
!         public void abort() {}
!     }
! 
!     private class BlockingRawAsyncEvent extends RawAsyncEvent
!             implements AsyncEvent.Blocking {
! 
!         BlockingRawAsyncEvent(RawEvent re) {
!             super(re);
!         }
      }
  
!     private class NonBlockingRawAsyncEvent extends RawAsyncEvent
!             implements AsyncEvent.NonBlocking {
  
          NonBlockingRawAsyncEvent(RawEvent re) {
!             super(re);
          }
      }
  
      /*
       * Register given event whose callback will be called once only.
       * (i.e. register new event for each callback)
       */
      public void registerEvent(RawEvent event) throws IOException {
!         if (event instanceof BlockingEvent) {
!             client.registerEvent(new BlockingRawAsyncEvent(event));
!         } else if (event instanceof NonBlockingEvent) {
!             client.registerEvent(new NonBlockingRawAsyncEvent(event));
!         } else {
              throw new InternalError();
          }
      }
  
      @Override
      public int read(ByteBuffer dst) throws IOException {
          return connection.read(dst);
      }
  
      @Override
      public boolean isOpen() {
--- 60,129 ----
      private class RawAsyncEvent extends AsyncEvent {
  
          private final RawEvent re;
  
          RawAsyncEvent(RawEvent re) {
+             super(AsyncEvent.BLOCKING); // BLOCKING & !REPEATING
+             this.re = re;
+         }
+ 
+         RawAsyncEvent(RawEvent re, int flags) {
+             super(flags);
              this.re = re;
          }
  
+         @Override
          public SelectableChannel channel() {
              return connection.channel();
          }
  
          // must return the selector interest op flags OR'd
+         @Override
          public int interestOps() {
              return re.interestOps();
          }
  
          // called when event occurs
+         @Override
          public void handle() {
              re.handle();
          }
  
!         @Override
!         public void abort() { }
      }
  
!     private class NonBlockingRawAsyncEvent extends RawAsyncEvent {
  
          NonBlockingRawAsyncEvent(RawEvent re) {
!             super(re, 0); // !BLOCKING & !REPEATING
          }
      }
  
      /*
       * Register given event whose callback will be called once only.
       * (i.e. register new event for each callback)
       */
      public void registerEvent(RawEvent event) throws IOException {
!         if (!(event instanceof NonBlockingEvent)) {
              throw new InternalError();
          }
+         if ((event.interestOps() & SelectionKey.OP_READ) != 0
+                 && connection.buffer.hasRemaining()) {
+             // FIXME: a hack to deal with leftovers from previous reads into an
+             // internal buffer (works in conjunction with change in
+             // java.net.http.PlainHttpConnection.readImpl(java.nio.ByteBuffer)
+             connection.channel().configureBlocking(false);
+             event.handle();
+         } else {
+             client.registerEvent(new NonBlockingRawAsyncEvent(event));
+         }
      }
  
      @Override
      public int read(ByteBuffer dst) throws IOException {
+         assert !connection.channel().isBlocking();
          return connection.read(dst);
      }
  
      @Override
      public boolean isOpen() {