java.nio.channels.WritePendingException#javax.websocket.CloseReason.CloseCodes源码实例Demo

/**
 * Test setting a maximum message length and make sure the stream is closed.
 *
 * @throws Exception in case of an error
 */
@Test
public void testMaxMessageLength() throws Exception {
    int maxMessageLength = 10;
    CountDownLatch latch = new CountDownLatch(1);
    EventStreamClosedClient eventListener = new EventStreamClosedClient(latch);

    try (EventStream streamEvents = new WebSocketEventStream(uri, new Token("token"), 0, 0,
            maxMessageLength, eventListener)) {
        latch.await(30, TimeUnit.SECONDS);
        Assert.assertTrue(streamEvents.isEventStreamClosed());
        Assert.assertEquals(CloseCodes.TOO_BIG.getCode(),
                eventListener.closeCode);
        String message = "Message length exceeded the configured maximum (" +
                maxMessageLength + " characters)";
        Assert.assertEquals(message, eventListener.closePhrase);
    }
}
 

@SuppressWarnings("unchecked")
protected void sendMessageBinary(ByteBuffer msg, boolean last) throws WsIOException {
    if (binaryMsgHandler instanceof WrappedMessageHandler) {
        long maxMessageSize = ((WrappedMessageHandler) binaryMsgHandler).getMaxMessageSize();
        if (maxMessageSize > -1 && msg.remaining() > maxMessageSize) {
            throw new WsIOException(new CloseReason(CloseCodes.TOO_BIG,
                    sm.getString("wsFrame.messageTooBig",
                            Long.valueOf(msg.remaining()),
                            Long.valueOf(maxMessageSize))));
        }
    }
    try {
        if (binaryMsgHandler instanceof MessageHandler.Partial<?>) {
            ((MessageHandler.Partial<ByteBuffer>) binaryMsgHandler).onMessage(msg, last);
        } else {
            // Caller ensures last == true if this branch is used
            ((MessageHandler.Whole<ByteBuffer>) binaryMsgHandler).onMessage(msg);
        }
    } catch (Throwable t) {
        handleThrowableOnSend(t);
    }
}
 

/**
 * Cleans up the resources still in use by WebSocket sessions created from
 * this container. This includes closing sessions and cancelling
 * {@link Future}s associated with blocking read/writes.
 */
public void destroy() {
    CloseReason cr = new CloseReason(
            CloseCodes.GOING_AWAY, sm.getString("wsWebSocketContainer.shutdown"));

    for (WsSession session : sessions.keySet()) {
        try {
            session.close(cr);
        } catch (IOException ioe) {
            log.debug(sm.getString(
                    "wsWebSocketContainer.sessionCloseFail", session.getId()), ioe);
        }
    }

    // Only unregister with AsyncChannelGroupUtil if this instance
    // registered with it
    if (asynchronousChannelGroup != null) {
        synchronized (asynchronousChannelGroupLock) {
            if (asynchronousChannelGroup != null) {
                AsyncChannelGroupUtil.unregister();
                asynchronousChannelGroup = null;
            }
        }
    }
}
 

@Test
public void testWsCloseThenTcpReset() throws Exception {
    startServer(TestEndpointConfig.class);

    TesterWsClient client = new TesterWsClient("localhost", getPort());
    client.httpUpgrade(BaseEndpointConfig.PATH);
    client.sendCloseFrame(CloseCodes.GOING_AWAY);
    client.forceCloseSocket();

    // WebSocket 1.1, section 2.1.5 requires this to be CLOSED_ABNORMALLY if
    // the container initiates the close and the close code from the client
    // if the client initiates it. When the client resets the TCP connection
    // after sending the close, different operating systems react different
    // ways. Some present the close message then drop the connection, some
    // just drop the connection. Therefore, this test has to handle both
    // close codes.
    awaitOnClose(CloseCodes.CLOSED_ABNORMALLY, CloseCodes.GOING_AWAY);
}
 

@Test
public void testTcpCloseInOnMessage() throws Exception {
    // TODO
    Assume.assumeFalse("This test currently fails for APR",
            getTomcatInstance().getConnector().getProtocolHandlerClassName().contains("Apr"));

    startServer(TestEndpointConfig.class);

    TesterWsClient client = new TesterWsClient("localhost", getPort());
    client.httpUpgrade(BaseEndpointConfig.PATH);
    client.sendTextMessage("Test");
    awaitLatch(events.onMessageCalled, "onMessage not called");

    client.closeSocket();
    events.onMessageWait.countDown();

    awaitOnClose(CloseCodes.CLOSED_ABNORMALLY);
}
 

@Test
public void testWsCloseThenTcpCloseWhenOnMessageSends() throws Exception {
    events.onMessageSends = true;

    startServer(TestEndpointConfig.class);

    TesterWsClient client = new TesterWsClient("localhost", getPort());
    client.httpUpgrade(BaseEndpointConfig.PATH);
    client.sendTextMessage("Test");
    awaitLatch(events.onMessageCalled, "onMessage not called");

    client.sendCloseFrame(CloseCodes.NORMAL_CLOSURE);
    client.closeSocket();
    events.onMessageWait.countDown();

    awaitOnClose(CloseCodes.CLOSED_ABNORMALLY);
}
 

@Test
public void testWsCloseThenTcpResetWhenOnMessageSends() throws Exception {
    events.onMessageSends = true;

    startServer(TestEndpointConfig.class);

    TesterWsClient client = new TesterWsClient("localhost", getPort());
    client.httpUpgrade(BaseEndpointConfig.PATH);
    client.sendTextMessage("Test");
    awaitLatch(events.onMessageCalled, "onMessage not called");

    client.sendCloseFrame(CloseCodes.NORMAL_CLOSURE);
    client.forceCloseSocket();
    events.onMessageWait.countDown();

    awaitOnClose(CloseCodes.CLOSED_ABNORMALLY);
}
 

@SuppressWarnings("unchecked")
private void sendMessageBinary(ByteBuffer msg, boolean last)
        throws WsIOException {
    if (binaryMsgHandler instanceof WrappedMessageHandler) {
        long maxMessageSize =
                ((WrappedMessageHandler) binaryMsgHandler).getMaxMessageSize();
        if (maxMessageSize > -1 && msg.remaining() > maxMessageSize) {
            throw new WsIOException(new CloseReason(CloseCodes.TOO_BIG,
                    sm.getString("wsFrame.messageTooBig",
                            Long.valueOf(msg.remaining()),
                            Long.valueOf(maxMessageSize))));
        }
    }
    try {
        if (binaryMsgHandler instanceof MessageHandler.Partial<?>) {
            ((MessageHandler.Partial<ByteBuffer>) binaryMsgHandler).onMessage(msg, last);
        } else {
            // Caller ensures last == true if this branch is used
            ((MessageHandler.Whole<ByteBuffer>) binaryMsgHandler).onMessage(msg);
        }
    } catch(Throwable t) {
        handleThrowableOnSend(t);
    }
}
 

/**
 * Cleans up the resources still in use by WebSocket sessions created from
 * this container. This includes closing sessions and cancelling
 * {@link Future}s associated with blocking read/writes.
 */
public void destroy() {
    CloseReason cr = new CloseReason(
            CloseCodes.GOING_AWAY, sm.getString("wsWebSocketContainer.shutdown"));

    for (WsSession session : sessions.keySet()) {
        try {
            session.close(cr);
        } catch (IOException ioe) {
            log.debug(sm.getString(
                    "wsWebSocketContainer.sessionCloseFail", session.getId()), ioe);
        }
    }

    // Only unregister with AsyncChannelGroupUtil if this instance
    // registered with it
    if (asynchronousChannelGroup != null) {
        synchronized (asynchronousChannelGroupLock) {
            if (asynchronousChannelGroup != null) {
                AsyncChannelGroupUtil.unregister();
                asynchronousChannelGroup = null;
            }
        }
    }
}
 

/**
    * When user leaves the activity.
    */
   @OnClose
   public void unregisterUser(Session websocket, CloseReason reason) {
Long toolSessionId = Long
	.valueOf(websocket.getRequestParameterMap().get(AttributeNames.PARAM_TOOL_SESSION_ID).get(0));
LearningWebsocketServer.websockets.get(toolSessionId).remove(websocket);

if (LearningWebsocketServer.log.isDebugEnabled()) {
    // If there was something wrong with the connection, put it into logs.
    LearningWebsocketServer.log.debug("User " + websocket.getUserPrincipal().getName()
	    + " left Mindmap with Tool Session ID: " + toolSessionId
	    + (!(reason.getCloseCode().equals(CloseCodes.GOING_AWAY)
		    || reason.getCloseCode().equals(CloseCodes.NORMAL_CLOSURE))
			    ? ". Abnormal close. Code: " + reason.getCloseCode() + ". Reason: "
				    + reason.getReasonPhrase()
			    : ""));
}
   }
 

/**
    * When user leaves the activity.
    */
   @OnClose
   public void unregisterUser(Session websocket, CloseReason reason) {
Long toolContentID = Long
	.valueOf(websocket.getRequestParameterMap().get(AttributeNames.PARAM_TOOL_CONTENT_ID).get(0));
websockets.get(toolContentID).remove(websocket);

if (log.isDebugEnabled()) {
    // If there was something wrong with the connection, put it into logs.
    log.debug("User " + websocket.getUserPrincipal().getName() + " left Dokumaran with Tool Content ID: "
	    + toolContentID
	    + (!(reason.getCloseCode().equals(CloseCodes.GOING_AWAY)
		    || reason.getCloseCode().equals(CloseCodes.NORMAL_CLOSURE))
			    ? ". Abnormal close. Code: " + reason.getCloseCode() + ". Reason: "
				    + reason.getReasonPhrase()
			    : ""));
}
   }
 

/**
    * When user leaves the activity.
    */
   @OnClose
   public void unregisterUser(Session websocket, CloseReason reason) {
Long toolSessionId = Long
	.valueOf(websocket.getRequestParameterMap().get(AttributeNames.PARAM_TOOL_SESSION_ID).get(0));
websockets.get(toolSessionId).remove(websocket);

if (log.isDebugEnabled()) {
    // If there was something wrong with the connection, put it into logs.
    log.debug("User " + websocket.getUserPrincipal().getName() + " left Leader Selection with Tool Session ID: "
	    + toolSessionId
	    + (!(reason.getCloseCode().equals(CloseCodes.GOING_AWAY)
		    || reason.getCloseCode().equals(CloseCodes.NORMAL_CLOSURE))
			    ? ". Abnormal close. Code: " + reason.getCloseCode() + ". Reason: "
				    + reason.getReasonPhrase()
			    : ""));
}
   }
 

/**
    * When user leaves the activity.
    */
   @OnClose
   public void unregisterUser(Session session, CloseReason reason) {
Long toolSessionId = Long
	.valueOf(session.getRequestParameterMap().get(AttributeNames.PARAM_TOOL_SESSION_ID).get(0));
Set<Websocket> sessionWebsockets = LearningWebsocketServer.websockets.get(toolSessionId);
Iterator<Websocket> websocketIterator = sessionWebsockets.iterator();
while (websocketIterator.hasNext()) {
    Websocket websocket = websocketIterator.next();
    if (websocket.session.equals(session)) {
	websocketIterator.remove();
	break;
    }
}

if (LearningWebsocketServer.log.isDebugEnabled()) {
    LearningWebsocketServer.log.debug(
	    "User " + session.getUserPrincipal().getName() + " left Chat with toolSessionId: " + toolSessionId
		    + (!(reason.getCloseCode().equals(CloseCodes.GOING_AWAY)
			    || reason.getCloseCode().equals(CloseCodes.NORMAL_CLOSURE))
				    ? ". Abnormal close. Code: " + reason.getCloseCode() + ". Reason: "
					    + reason.getReasonPhrase()
				    : ""));
}
   }
 

/**
    * When user leaves the activity.
    */
   @OnClose
   public void unregisterUser(Session websocket, CloseReason reason) {
Long toolSessionId = Long
	.valueOf(websocket.getRequestParameterMap().get(AttributeNames.PARAM_TOOL_SESSION_ID).get(0));
LearningWebsocketServer.websockets.get(toolSessionId).remove(websocket);

if (log.isDebugEnabled()) {
    // If there was something wrong with the connection, put it into logs.
    log.debug("User " + websocket.getUserPrincipal().getName() + " left Scratchie with Tool Session ID: "
	    + toolSessionId
	    + (!(reason.getCloseCode().equals(CloseCodes.GOING_AWAY)
		    || reason.getCloseCode().equals(CloseCodes.NORMAL_CLOSURE))
			    ? ". Abnormal close. Code: " + reason.getCloseCode() + ". Reason: "
				    + reason.getReasonPhrase()
			    : ""));
}
   }
 

/**
    * When user leaves the activity.
    */
   @OnClose
   public void unregisterUser(Session websocket, CloseReason reason) {
Long toolSessionId = Long
	.valueOf(websocket.getRequestParameterMap().get(AttributeNames.PARAM_TOOL_SESSION_ID).get(0));
LearningWebsocketServer.websockets.get(toolSessionId).remove(websocket);

if (LearningWebsocketServer.log.isDebugEnabled()) {
    // If there was something wrong with the connection, put it into logs.
    LearningWebsocketServer.log.debug("User " + websocket.getUserPrincipal().getName()
	    + " left Scribe with Tool Session ID: " + toolSessionId
	    + (!(reason.getCloseCode().equals(CloseCodes.GOING_AWAY)
		    || reason.getCloseCode().equals(CloseCodes.NORMAL_CLOSURE))
			    ? ". Abnormal close. Code: " + reason.getCloseCode() + ". Reason: "
				    + reason.getReasonPhrase()
			    : ""));
}
   }
 

/**
 * Close the connection
 */
@Override
public void close() {
	if(userSession == null) {
		return;
	}
	
	try {
		final CloseReason closeReason = new CloseReason(CloseCodes.NORMAL_CLOSURE, "Socket closed");
		userSession.close(closeReason);
	} catch (Throwable e) {
		logger.error("Got exception while closing socket", e);
	}
	
	userSession = null;
}
 

@Timed(name = "websocket_onError_timer",
    reusable = true,
    tags = "label=websocket")
@Counted(name = "websocket_onError_count",
    monotonic = true,
    reusable = true,
    tags = "label=websocket")
@Metered(name = "websocket_onError_meter",
    reusable = true,
    tags = "label=websocket")
@OnError
public void onError(Session session, Throwable t) {
    Log.log(Level.FINE, this, "A problem occurred on connection", t);

    // TODO: Careful with what might revealed about implementation details!!
    // We're opting for making debug easy..
    tryToClose(session,
            new CloseReason(CloseReason.CloseCodes.UNEXPECTED_CONDITION,
                    trimReason(t.getClass().getName())));
}
 

/**
 * Try sending the {@link Message} using
 * {@link Session#getBasicRemote()}, {@link Basic#sendObject(Object)}.
 *
 * @param session Session to send the message on
 * @param message Message to send
 * @return true if send was successful, or false if it failed
 */
private boolean sendMessageToSession(Session session, Message message) {
    if (session.isOpen()) {
        try {
            session.getBasicRemote().sendObject(message);
            return true;
        } catch (EncodeException e) {
            // Something was wrong encoding this message, but the connection
            // is likely just fine.
            Log.log(Level.FINE, this, "Unexpected condition writing message", e);
        } catch (IOException ioe) {
            // An IOException, on the other hand, suggests the connection is
            // in a bad state.
            Log.log(Level.FINE, this, "Unexpected condition writing message", ioe);
            tryToClose(session, new CloseReason(CloseCodes.UNEXPECTED_CONDITION, trimReason(ioe.toString())));
        }
    }
    return false;
}
 

@Test
public void testWsCloseThenTcpResetWhenOnMessageSends() throws Exception {
    events.onMessageSends = true;

    startServer(TestEndpointConfig.class);

    TesterWsCloseClient client = new TesterWsCloseClient("localhost", getPort());
    client.httpUpgrade(BaseEndpointConfig.PATH);
    client.sendMessage("Test");
    awaitLatch(events.onMessageCalled, "onMessage not called");

    client.sendCloseFrame(CloseCodes.NORMAL_CLOSURE);
    client.forceCloseSocket();
    events.onMessageWait.countDown();

    // APR will see close from client before it sees the TCP reset
    awaitOnClose(CloseCodes.CLOSED_ABNORMALLY, CloseCodes.NORMAL_CLOSURE);
}
 

@Override
public void shutdown() {
    synchronized (connections) {
        parent.channelCloses(this);
        for (final Session connection : connections) {
            final ExecutorService executorService = connectionThreads.get(connection);
            try {
                // If there is no executor service for a client, it may has already been shut down.
                if (executorService != null) {
                    executorService.shutdownNow();
                }
                connection
                        .close(new CloseReason(CloseCodes.GOING_AWAY, "This SynchronizeFX channel is closed now."));
            } catch (final IOException e) {
                callback.onClientConnectionError(connection,
                        new SynchronizeFXException("Failed to close the connection to a connected client.", e));
            } finally {
                connectionThreads.remove(connection);
            }
        }
        connections.clear();
    }
    callback = null;
}
 

private void handleSendFailureWithEncode(Throwable t) throws IOException, EncodeException {
    // First, unwrap any execution exception
    if (t instanceof ExecutionException) {
        t = t.getCause();
    }

    // Close the session
    wsSession.doClose(new CloseReason(CloseCodes.GOING_AWAY, t.getMessage()),
            new CloseReason(CloseCodes.CLOSED_ABNORMALLY, t.getMessage()));

    // Rethrow the exception
    if (t instanceof EncodeException) {
        throw (EncodeException) t;
    }
    if (t instanceof IOException) {
        throw (IOException) t;
    }
    throw new IOException(t);
}
 

/**
 * Cleans up the resources still in use by WebSocket sessions created from
 * this container. This includes closing sessions and cancelling
 * {@link Future}s associated with blocking read/writes.
 */
public void destroy() {
    CloseReason cr = new CloseReason(
            CloseCodes.GOING_AWAY, sm.getString("wsWebSocketContainer.shutdown"));

    for (WsSession session : sessions.keySet()) {
        try {
            session.close(cr);
        } catch (IOException ioe) {
            log.debug(sm.getString(
                    "wsWebSocketContainer.sessionCloseFail", session.getId()), ioe);
        }
    }

    // Only unregister with AsyncChannelGroupUtil if this instance
    // registered with it
    if (asynchronousChannelGroup != null) {
        synchronized (asynchronousChannelGroupLock) {
            if (asynchronousChannelGroup != null) {
                AsyncChannelGroupUtil.unregister();
                asynchronousChannelGroup = null;
            }
        }
    }
}
 

@Test
public void testWsCloseThenTcpReset() throws Exception {
    startServer(TestEndpointConfig.class);

    TesterWsCloseClient client = new TesterWsCloseClient("localhost", getPort());
    client.httpUpgrade(BaseEndpointConfig.PATH);
    client.sendCloseFrame(CloseCodes.GOING_AWAY);
    client.forceCloseSocket();

    // WebSocket 1.1, section 2.1.5 requires this to be CLOSED_ABNORMALLY if
    // the container initiates the close and the close code from the client
    // if the client initiates it. When the client resets the TCP connection
    // after sending the close, different operating systems react different
    // ways. Some present the close message then drop the connection, some
    // just drop the connection. Therefore, this test has to handle both
    // close codes.
    awaitOnClose(CloseCodes.CLOSED_ABNORMALLY, CloseCodes.GOING_AWAY);
}
 

@Test
public void testWsCloseThenTcpCloseWhenOnMessageSends() throws Exception {
    events.onMessageSends = true;

    startServer(TestEndpointConfig.class);

    TesterWsCloseClient client = new TesterWsCloseClient("localhost", getPort());
    client.httpUpgrade(BaseEndpointConfig.PATH);
    client.sendMessage("Test");
    awaitLatch(events.onMessageCalled, "onMessage not called");

    client.sendCloseFrame(CloseCodes.NORMAL_CLOSURE);
    client.closeSocket();
    events.onMessageWait.countDown();

    // BIO will see close from client before it sees the TCP close
    awaitOnClose(CloseCodes.CLOSED_ABNORMALLY, CloseCodes.NORMAL_CLOSURE);
}
 

protected void sendMessage(String msg) {
    try {
        session.getBasicRemote().sendText(msg);
    } catch (IOException ioe) {
        CloseReason cr =
                new CloseReason(CloseCodes.CLOSED_ABNORMALLY, ioe.getMessage());
        try {
            session.close(cr);
        } catch (IOException ioe2) {
            // Ignore
        }
    }
}
 

private void sendCloseMessage(CloseReason closeReason) {
    // 125 is maximum size for the payload of a control message
    ByteBuffer msg = ByteBuffer.allocate(125);
    CloseCode closeCode = closeReason.getCloseCode();
    // CLOSED_ABNORMALLY should not be put on the wire
    if (closeCode == CloseCodes.CLOSED_ABNORMALLY) {
        // PROTOCOL_ERROR is probably better than GOING_AWAY here
        msg.putShort((short) CloseCodes.PROTOCOL_ERROR.getCode());
    } else {
        msg.putShort((short) closeCode.getCode());
    }

    String reason = closeReason.getReasonPhrase();
    if (reason != null && reason.length() > 0) {
        appendCloseReasonWithTruncation(msg, reason);
    }
    msg.flip();
    try {
        wsRemoteEndpoint.sendMessageBlock(Constants.OPCODE_CLOSE, msg, true);
    } catch (IOException | WritePendingException e) {
        // Failed to send close message. Close the socket and let the caller
        // deal with the Exception
        if (log.isDebugEnabled()) {
            log.debug(sm.getString("wsSession.sendCloseFail", id), e);
        }
        wsRemoteEndpoint.close();
        // Failure to send a close message is not unexpected in the case of
        // an abnormal closure (usually triggered by a failure to read/write
        // from/to the client. In this case do not trigger the endpoint's
        // error handling
        if (closeCode != CloseCodes.CLOSED_ABNORMALLY) {
            localEndpoint.onError(this, e);
        }
    } finally {
        webSocketContainer.unregisterSession(getSessionMapKey(), this);
    }
}
 

protected void checkExpiration() {
    long timeout = maxIdleTimeout;
    if (timeout < 1) {
        return;
    }

    if (System.currentTimeMillis() - lastActive > timeout) {
        String msg = sm.getString("wsSession.timeout", getId());
        if (log.isDebugEnabled()) {
            log.debug(msg);
        }
        doClose(new CloseReason(CloseCodes.GOING_AWAY, msg),
                new CloseReason(CloseCodes.CLOSED_ABNORMALLY, msg));
    }
}
 

@SuppressWarnings("unchecked")
protected void sendMessageText(boolean last) throws WsIOException {
    if (textMsgHandler instanceof WrappedMessageHandler) {
        long maxMessageSize = ((WrappedMessageHandler) textMsgHandler).getMaxMessageSize();
        if (maxMessageSize > -1 && messageBufferText.remaining() > maxMessageSize) {
            throw new WsIOException(new CloseReason(CloseCodes.TOO_BIG,
                    sm.getString("wsFrame.messageTooBig",
                            Long.valueOf(messageBufferText.remaining()),
                            Long.valueOf(maxMessageSize))));
        }
    }

    try {
        if (textMsgHandler instanceof MessageHandler.Partial<?>) {
            ((MessageHandler.Partial<String>) textMsgHandler)
                    .onMessage(messageBufferText.toString(), last);
        } else {
            // Caller ensures last == true if this branch is used
            ((MessageHandler.Whole<String>) textMsgHandler)
                    .onMessage(messageBufferText.toString());
        }
    } catch (Throwable t) {
        handleThrowableOnSend(t);
    } finally {
        messageBufferText.clear();
    }
}
 

private void handleThrowableOnSend(Throwable t) throws WsIOException {
    ExceptionUtils.handleThrowable(t);
    wsSession.getLocal().onError(wsSession, t);
    CloseReason cr = new CloseReason(CloseCodes.CLOSED_ABNORMALLY,
            sm.getString("wsFrame.ioeTriggeredClose"));
    throw new WsIOException(cr);
}
 

protected void sendMessage(String msg) {
    try {
        session.getBasicRemote().sendText(msg);
    } catch (IOException ioe) {
        CloseReason cr =
                new CloseReason(CloseCodes.CLOSED_ABNORMALLY, ioe.getMessage());
        try {
            session.close(cr);
        } catch (IOException ioe2) {
            // Ignore
        }
    }
}