java.nio.channels.ClosedSelectorException#java.nio.channels.ClosedChannelException源码实例Demo

@Test(expected = TransportClosedException.class)
public void shouldNotifyTransportObserverIfAnClosedChannelExceptionIsThrownWhileWritingACollection() throws Exception
{
    mockery.checking(new Expectations()
    {
        {
            //when
            one(writableByteChannel).write(with(any(ByteBuffer.class)));
            will(throwException(new ClosedChannelException()));

            //then
            one(connectionObserver).connectionClosed();
        }
    });

    handler.send(Arrays.asList(new FixMessageBuilder().build()));

}
 

@Override
public <T> SctpChannel setOption(SctpSocketOption<T> name, T value)
        throws IOException {
    if (name == null)
        throw new NullPointerException();
    if (!supportedOptions().contains(name))
        throw new UnsupportedOperationException("'" + name + "' not supported");

    synchronized (stateLock) {
        if (!isOpen())
            throw new ClosedChannelException();

        SctpNet.setSocketOption(fdVal, name, value, 0 /*oneToOne*/);
    }
    return this;
}
 

/**
 * Adjusts this channel's blocking mode.
 *
 * <p> If the given blocking mode is different from the current blocking
 * mode then this method invokes the {@link #implConfigureBlocking
 * implConfigureBlocking} method, while holding the appropriate locks, in
 * order to change the mode.  </p>
 */
public final SelectableChannel configureBlocking(boolean block)
    throws IOException
{
    synchronized (regLock) {
        if (!isOpen())
            throw new ClosedChannelException();
        boolean blocking = !nonBlocking;
        if (block != blocking) {
            if (block && haveValidKeys())
                throw new IllegalBlockingModeException();
            implConfigureBlocking(block);
            nonBlocking = !block;
        }
    }
    return this;
}
 

@Override
public Mono<DuplexConnection> connect() {
  return Mono.defer(
      () ->
          now() < nextConnectPermitMillis
              ? Mono.error(new ClosedChannelException())
              : clientTransport
                  .connect()
                  .map(
                      c -> {
                        if (curConnection.compareAndSet(null, c)) {
                          return c;
                        } else {
                          throw new IllegalStateException(
                              "Transport supports at most 1 connection");
                        }
                      }));
}
 

public SelectionKey registerForRead(
    final ReceiveChannelEndpoint channelEndpoint, final UdpChannelTransport transport, final int transportIndex)
{
    SelectionKey key = null;
    try
    {
        final ChannelAndTransport channelAndTransport = new ChannelAndTransport(
            channelEndpoint, transport, transportIndex);

        key = transport.receiveDatagramChannel().register(selector, SelectionKey.OP_READ, channelAndTransport);
        channelAndTransports = ArrayUtil.add(channelAndTransports, channelAndTransport);
    }
    catch (final ClosedChannelException ex)
    {
        LangUtil.rethrowUnchecked(ex);
    }

    return key;
}
 

protected void addNewClient(ClientProcessData clientData) {
  SelectionKey clientKey;
  try {
    // if client is already in execution then register OP_WRITE interest
    // too, else only OP_READ
    /*if (clientData.idle) {
      clientKey = clientData.clientSocket.registerSelector(this.selector,
          SelectionKey.OP_READ);
    }
    else*/ {
      clientKey = clientData.clientSocket.registerSelector(this.selector,
          SelectionKey.OP_READ | SelectionKey.OP_WRITE);
    }

    clientData.key = clientKey;
    clientKey.attach(clientData);
  } catch (ClosedChannelException cce) {
    cleanupSelectionKey(clientData);
  } catch (IOException ioe) {
    LOGGER.warn("Failed to register accepted connection to selector!",
        ioe);
    cleanupSelectionKey(clientData);
  }
}
 

public void testNewInputStreamReadableByteChannel() throws Exception {
    ByteBuffer readbcbuf = ByteBuffer.allocateDirect(this.testNum);
    byte[] readbuf = new byte[this.testNum];
    this.fins = new FileInputStream(tmpFile);
    ReadableByteChannel readbc = this.fins.getChannel();
    assertEquals(this.fileSize, this.fins.available());
    assertTrue(readbc.isOpen());
    InputStream testins = Channels.newInputStream(readbc);
    // read in testins and fins use the same pointer
    testins.read(readbuf);
    assertEquals(this.fins.available(), this.fileSize - this.testNum);
    int readNum = readbc.read(readbcbuf);
    assertEquals(readNum, this.testNum);
    assertEquals(this.fins.available(), this.fileSize - this.testNum * 2);
    testins.read(readbuf);
    assertEquals(this.fins.available(), this.fileSize - this.testNum * 3);
    // readbc.close() affect testins
    readbc.close();
    assertFalse(readbc.isOpen());
    try {
        testins.read(readbuf);
        fail();
    } catch (ClosedChannelException e) {
        // correct
    }
}
 

public int setBytes(int index, ScatteringByteChannel in, int length) throws IOException {
    ByteBuffer buf = ByteBuffer.wrap(array, index, length);
    int readBytes = 0;

    do {
        int localReadBytes;
        try {
            localReadBytes = in.read(buf);
        } catch (ClosedChannelException e) {
            localReadBytes = -1;
        }
        if (localReadBytes < 0) {
            if (readBytes == 0) {
                return -1;
            } else {
                break;
            }
        } else if (localReadBytes == 0) {
            break;
        }
        readBytes += localReadBytes;
    } while (readBytes < length);

    return readBytes;
}
 

@Override
public <T> SctpMultiChannel setOption(SctpSocketOption<T> name,
                                      T value,
                                      Association association)
        throws IOException {
    if (name == null)
        throw new NullPointerException();
    if (!(supportedOptions().contains(name)))
        throw new UnsupportedOperationException("'" + name + "' not supported");

    synchronized (stateLock) {
        if (association != null && (name.equals(SCTP_PRIMARY_ADDR) ||
                name.equals(SCTP_SET_PEER_PRIMARY_ADDR))) {
            checkAssociation(association);
        }
        if (!isOpen())
            throw new ClosedChannelException();

        int assocId = association == null ? 0 : association.associationID();
        SctpNet.setSocketOption(fdVal, name, value, assocId);
    }
    return this;
}
 

/**
 * Returns a future holding the serialized request result.
 *
 * <p>If the channel has been established, forward the call to the
 * established channel, otherwise queue it for when the channel is
 * handed in.
 *
 * @param request the request to be sent.
 * @return Future holding the serialized result
 */
CompletableFuture<RESP> sendRequest(REQ request) {
	synchronized (connectLock) {
		if (failureCause != null) {
			return FutureUtils.getFailedFuture(failureCause);
		} else if (connectionShutdownFuture.get() != null) {
			return FutureUtils.getFailedFuture(new ClosedChannelException());
		} else {
			if (established != null) {
				return established.sendRequest(request);
			} else {
				// Queue this and handle when connected
				final PendingRequest pending = new PendingRequest(request);
				queuedRequests.add(pending);
				return pending;
			}
		}
	}
}
 

@Override
@SuppressWarnings("unchecked")
public <T> T getOption(SctpSocketOption<T> name, Association association)
        throws IOException {
    if (name == null)
        throw new NullPointerException();
    if (!supportedOptions().contains(name))
        throw new UnsupportedOperationException("'" + name + "' not supported");

    synchronized (stateLock) {
        if (association != null && (name.equals(SCTP_PRIMARY_ADDR) ||
                name.equals(SCTP_SET_PEER_PRIMARY_ADDR))) {
            checkAssociation(association);
        }
        if (!isOpen())
            throw new ClosedChannelException();

        int assocId = association == null ? 0 : association.associationID();
        return (T)SctpNet.getSocketOption(fdVal, name, assocId);
    }
}
 

@Override
public SctpChannel branch(Association association)
        throws IOException {
    synchronized (stateLock) {
        checkAssociation(association);
        if (!isOpen())
            throw new ClosedChannelException();

        FileDescriptor bFd = SctpNet.branch(fdVal,
                                            association.associationID());
        /* successfully branched, we can now remove it from assoc list */
        removeAssociation(association);

        return new SctpChannelImpl(provider(), bFd, association);
    }
}
 

@Test
public void testOversizedRequest() {
    EmbeddedChannel embedder = new EmbeddedChannel(new HttpObjectAggregator(4));
    HttpRequest message = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.PUT, "http://localhost");
    HttpContent chunk1 = new DefaultHttpContent(Unpooled.copiedBuffer("test", CharsetUtil.US_ASCII));
    HttpContent chunk2 = new DefaultHttpContent(Unpooled.copiedBuffer("test2", CharsetUtil.US_ASCII));
    HttpContent chunk3 = LastHttpContent.EMPTY_LAST_CONTENT;

    assertFalse(embedder.writeInbound(message));
    assertFalse(embedder.writeInbound(chunk1));
    assertFalse(embedder.writeInbound(chunk2));

    FullHttpResponse response = embedder.readOutbound();
    assertEquals(HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, response.status());
    assertEquals("0", response.headers().get(HttpHeaderNames.CONTENT_LENGTH));
    assertFalse(embedder.isOpen());

    try {
        assertFalse(embedder.writeInbound(chunk3));
        fail();
    } catch (Exception e) {
        assertTrue(e instanceof ClosedChannelException);
    }

    assertFalse(embedder.finish());
}
 

public int setBytes(int index, ScatteringByteChannel in, int length) throws IOException {
    ByteBuffer buf = ByteBuffer.wrap(array, index, length);
    int readBytes = 0;

    do {
        int localReadBytes;
        try {
            localReadBytes = in.read(buf);
        } catch (ClosedChannelException e) {
            localReadBytes = -1;
        }
        if (localReadBytes < 0) {
            if (readBytes == 0) {
                return -1;
            } else {
                break;
            }
        } else if (localReadBytes == 0) {
            break;
        }
        readBytes += localReadBytes;
    } while (readBytes < length);

    return readBytes;
}
 

@Override
@SuppressWarnings("unchecked")
public <T> T getOption(SctpSocketOption<T> name, Association association)
        throws IOException {
    if (name == null)
        throw new NullPointerException();
    if (!supportedOptions().contains(name))
        throw new UnsupportedOperationException("'" + name + "' not supported");

    synchronized (stateLock) {
        if (association != null && (name.equals(SCTP_PRIMARY_ADDR) ||
                name.equals(SCTP_SET_PEER_PRIMARY_ADDR))) {
            checkAssociation(association);
        }
        if (!isOpen())
            throw new ClosedChannelException();

        int assocId = association == null ? 0 : association.associationID();
        return (T)SctpNet.getSocketOption(fdVal, name, assocId);
    }
}
 

@Override
public ChannelOutput receive() throws IOException {
  if (!connected) {
    throw new ClosedChannelException();
  }

  // consume the size header and return the remaining response.
  ByteBuffer streamSizeBuffer = ByteBuffer.allocate(8);
  while (streamSizeBuffer.position() < streamSizeBuffer.capacity()) {
    int read = readChannel.read();
    if (read == -1) {
      throw new IOException("Could not read complete size from readChannel ");
    }
    streamSizeBuffer.put((byte) read);
  }
  streamSizeBuffer.flip();
  return new ChannelOutput(readChannel, streamSizeBuffer.getLong() - 8);
}
 

@Override
public <T> SctpServerChannel setOption(SctpSocketOption<T> name, T value)
        throws IOException {
    if (name == null)
        throw new NullPointerException();
    if (!supportedOptions().contains(name))
        throw new UnsupportedOperationException("'" + name + "' not supported");

    synchronized (stateLock) {
        if (!isOpen())
            throw new ClosedChannelException();

        SctpNet.setSocketOption(fdVal, name, value, 0 /*oneToOne*/);
        return this;
    }
}
 

@Test
public void chunkedWithoutFinalCRLF() throws Exception {
    encodedResponse.set("HTTP/1.1 200 OK\r\n" +
            "Content-Type: text/plain\r\n" +
            "Transfer-Encoding: chunked\r\n" +
            "Connection: close\r\n" + "\r\n" +
            "5\r\n" +
            "hello\r\n" +
            "0\r\n");   // no final CRLF

    HttpRequest request = client.get("/");
    ReservedBlockingHttpConnection connection = client.reserveConnection(request);
    // Wait until a server closes the connection:
    connection.connectionContext().onClose().whenFinally(connectionClosedLatch::countDown).subscribe();

    assertThrows(ClosedChannelException.class, () -> connection.request(request));
    connectionClosedLatch.await();
}
 

@Override
public Future<Long> readInto(AsyncWritableChannel asyncWritableChannel, Callback<Long> callback) {
  Future<Long> future;
  if (!channelOpen.get()) {
    ClosedChannelException closedChannelException = new ClosedChannelException();
    FutureResult<Long> futureResult = new FutureResult<Long>();
    futureResult.done(0L, closedChannelException);
    future = futureResult;
    if (callback != null) {
      callback.onCompletion(0L, closedChannelException);
    }
  } else if (!channelEmptied.compareAndSet(false, true)) {
    throw new IllegalStateException("ReadableStreamChannel cannot be read more than once");
  } else {
    future = asyncWritableChannel.write(buffer, callback);
  }
  return future;
}
 

private void handleReadWriteIoException(IOException e) {
    if (e instanceof ClosedChannelException && !tcpNioTransport.isConnected()) {
        // The connection is already closed.
        return;
    }

   connectionInternal.notifyConnectionError(e);
}
 

@Override
@SuppressWarnings("unchecked")
public <T> T getOption(SctpSocketOption<T> name) throws IOException {
    if (name == null)
        throw new NullPointerException();
    if (!supportedOptions().contains(name))
        throw new UnsupportedOperationException("'" + name + "' not supported");

    synchronized (stateLock) {
        if (!isOpen())
            throw new ClosedChannelException();

        return (T)SctpNet.getSocketOption(fdVal, name, 0 /*oneToOne*/);
    }
}
 

private void ensureOpenAndUnconnected() throws IOException {
    synchronized (stateLock) {
        if (!isOpen())
            throw new ClosedChannelException();
        if (isConnected())
            throw new AlreadyConnectedException();
        if (state == ChannelState.PENDING)
            throw new ConnectionPendingException();
    }
}
 

/**
 * Check the integrity of all the nodes in the tree. Calls
 * {@link #assertNodeIntegrity} for every node in the tree.
 */
public void assertIntegrity() {
    try {
        for (int i = 0; i < getNodeCount(); i++) {
            assertNodeIntegrity(getNode(i));
        }
    } catch (ClosedChannelException e) {
        fail(e.getMessage());
    }
}
 

private void assert_select_OP_ACCEPT(SelectType type, int timeout)
        throws IOException, ClosedChannelException {
    SocketChannel sc = SocketChannel.open();
    SocketChannel client = null;
    try {
        ssc.register(selector, SelectionKey.OP_ACCEPT);
        sc.connect(localAddress);
        int count = blockingSelect(type, timeout);
        assertEquals(1, count);
        Set<SelectionKey> selectedKeys = selector.selectedKeys();
        assertEquals(1, selectedKeys.size());
        SelectionKey key = selectedKeys.iterator().next();
        assertEquals(ssc.keyFor(selector), key);
        assertEquals(SelectionKey.OP_ACCEPT, key.readyOps());
        // select again, it should return 0
        count = selectOnce(type, timeout);
        assertEquals(0,count);
        // but selectedKeys remains the same as previous
        assertSame(selectedKeys, selector.selectedKeys());
        client = ssc.accept();
        selectedKeys.clear();
    } finally {
        try {
            sc.close();
        } catch (IOException e) {
            // do nothing
        }
        if (null != client) {
            client.close();
        }
    }
    ssc.keyFor(selector).cancel();
}
 

private void postRequest(final Request request) throws AgentControlChannelException {
    if (_link != null) {
        try {
            _link.send(request.toBytes());
        } catch (final ClosedChannelException e) {
            s_logger.warn("Unable to post agent control reques: " + request.toString());
            throw new AgentControlChannelException("Unable to post agent control request due to " + e.getMessage());
        }
    } else {
        throw new AgentControlChannelException("Unable to post agent control request as link is not available");
    }
}
 

/**
 * Closes the channel and resolves all pending chunks with a {@link ClosedChannelException}. Also queues a poison
 * so that {@link #getNextChunk()} starts returning {@code null}.
 */
@Override
public void close() {
  if (channelOpen.compareAndSet(true, false)) {
    resolveAllRemainingChunks(new ClosedChannelException());
  }
}
 

/**
 * From http://vaibhavblogs.org/2012/12/common-java-networking-exceptions/
 */
public static boolean isNetworkException(Throwable e) {
  return e instanceof SocketException || e instanceof ClosedChannelException ||
      e instanceof InterruptedIOException || e instanceof ProtocolException ||
      e instanceof SSLException || e instanceof UnknownHostException ||
      e instanceof UnknownServiceException;
}
 

private void handleDeferredRegistrations()
{
    synchronized (deferredRegistrations) {
        int deferredListSize = deferredRegistrations.size();
        for (int i = 0; i < deferredListSize; i++) {
            EventHandler eventHandler =
                (EventHandler)deferredRegistrations.get(i);
            if (orb.transportDebugFlag) {
                dprint(".handleDeferredRegistrations: " + eventHandler);
            }
            SelectableChannel channel = eventHandler.getChannel();
            SelectionKey selectionKey = null;
            try {
                selectionKey =
                    channel.register(selector,
                                     eventHandler.getInterestOps(),
                                     (Object)eventHandler);
            } catch (ClosedChannelException e) {
                if (orb.transportDebugFlag) {
                    dprint(".handleDeferredRegistrations: " + e);
                }
            }
            eventHandler.setSelectionKey(selectionKey);
        }
        deferredRegistrations.clear();
    }
}
 

@Test(expected = IOException.class)
public void testConcurrentClose() throws Throwable {
  FileChannel spyChannel = spy(FileChannel.class);
  MappedPageSource source = new MappedPageSource(dataFile) {
    @Override
    public FileChannel getReadableChannel() {
      return spyChannel;
    }
  };
  FileBackedStorageEngine<byte[], byte[]> engine = new FileBackedStorageEngine<>(source, Long.MAX_VALUE, MemoryUnit.BYTES, PersistentByteArrayPortability.INSTANCE, PersistentByteArrayPortability.INSTANCE);
  when(spyChannel.read(notNull(), anyLong()))
    .thenAnswer(o -> {
      //deterministically simulate another thread closing the engine while
      //this thread is reading.
      engine.close();
      throw new ClosedChannelException();
    });
  try {
    byte[] buffer = new byte[10];
    long p = engine.writeMapping(new byte[0], buffer, 0, 0);
    Assert.assertTrue(p >= 0);
    engine.flush();
    engine.readValue(0);
  } catch (Throwable e) {
    Throwable cause = e;
    while (cause.getCause() != null) {
      cause = cause.getCause();
    }
    throw cause;
  } finally {
    source.close();
  }
}
 

@Test
public void testGetOptionWhenClosed() {
    ch.close().syncUninterruptibly();
    try {
        ch.config().getSoLinger();
        fail();
    } catch (ChannelException e) {
        assertTrue(e.getCause() instanceof ClosedChannelException);
    }
}