io.netty.handler.codec.http.websocketx.PongWebSocketFrame #javax.websocket.Session源码实例Demo

@Timed(name = "websocket_onOpen_timer",
    reusable = true,
    tags = "label=websocket")
@Counted(name = "websocket_onOpen_count",
    monotonic = true,
    reusable = true,
    tags = "label=websocket")
@Metered(name = "websocket_onOpen_meter",
    reusable = true,
    tags = "label=websocket")
@OnOpen
public void onOpen(Session session, EndpointConfig ec) {
    Log.log(Level.FINE, this, "A new connection has been made to the room.");

    // All we have to do in onOpen is send the acknowledgement
    sendMessage(session, Message.ACK_MSG);
}
 

@Override
public void onOpen(Session session, EndpointConfig config) {
    this.session = session;
    try {
        Thread.sleep(100);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    session.addMessageHandler(new MessageHandler.Whole<String>() {

        @Override
        public void onMessage(String message) {
            responses.add(message);
        }
    });
}
 

@SuppressWarnings("unchecked")
<T extends WebSocketMessage> void onMessage(
    final Session session, final MessageEnvelope envelope) {
  determineMatchingMessageType(envelope)
      .ifPresent(
          messageType -> {
            final T payload = (T) envelope.getPayload(messageType.getPayloadType());

            getListenersForMessageTypeId(envelope.getMessageTypeId())
                .map(messageListener -> (MessageListener<T>) messageListener)
                .forEach(
                    messageListener ->
                        messageListener.listener.accept(
                            WebSocketMessageContext.<T>builder()
                                .messagingBus(this)
                                .senderSession(session)
                                .message(payload)
                                .build()));
          });
}
 

@OnMessage
public void echoTextMessage(Session session, String msg, boolean last) {
    if (sb == null) {
        sb = new StringBuilder();
    }
    sb.append(msg);
    if (last) {
        // Before we send the next message, have to wait for the previous
        // message to complete
        try {
            f.get();
        } catch (InterruptedException | ExecutionException e) {
            // Let the container deal with it
            throw new RuntimeException(e);
        }
        f = session.getAsyncRemote().sendText(sb.toString());
        sb = null;
    }
}
 

private void disconnectSession(
    final Collection<Session> sessions, final String disconnectMessage) {
  // Do session disconnects as its own step to avoid concurrent modification.
  sessions.forEach(
      session -> {
        try {
          session.close(
              new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, disconnectMessage));
        } catch (final IOException e) {
          log.warn(
              "While closing session, "
                  + "session close threw an exception, session is left open? {}",
              session.isOpen(),
              e);
        }
      });
}
 

/**
    * 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()
			    : ""));
}
   }
 

@Override
public void onClose(Session session, CloseReason closeReason) {
    Room room = getRoom(false);
    if (room != null) {
        room.invokeAndWait(new Runnable() {
            @Override
            public void run() {
                try {
                    // Player can be null if it couldn't enter the room
                    if (player != null) {
                        // Remove this player from the room.
                        player.removeFromRoom();

                        // Set player to null to prevent NPEs when onMessage events
                        // are processed (from other threads) after onClose has been
                        // called from different thread which closed the Websocket session.
                        player = null;
                    }
                } catch (RuntimeException ex) {
                    log.error("Unexpected exception: " + ex.toString(), ex);
                }
            }
        });
    }
}
 

public Session connectToServer(Class<?> aClass, WebsocketConnectionBuilder connectionBuilder) throws DeploymentException, IOException {
    if (closed) {
        throw new ClosedChannelException();
    }
    ConfiguredClientEndpoint config = getClientEndpoint(aClass, true);
    if (config == null) {
        throw JsrWebSocketMessages.MESSAGES.notAValidClientEndpointType(aClass);
    }
    try {
        AnnotatedEndpointFactory factory = config.getFactory();
        InstanceHandle<?> instance = config.getInstanceFactory().createInstance();
        return connectToServerInternal(factory.createInstance(instance), config, connectionBuilder);
    } catch (InstantiationException e) {
        throw new RuntimeException(e);
    }
}
 

/**
    * Removes Learner websocket from the collection.
    */
   @OnClose
   public void unregisterUser(Session session, CloseReason reason) {
String login = session.getUserPrincipal().getName();
if (login == null) {
    return;
}

Long lessonId = Long.valueOf(session.getRequestParameterMap().get(AttributeNames.PARAM_LESSON_ID).get(0));
Map<String, Session> lessonWebsockets = CommandWebsocketServer.websockets.get(lessonId);
if (lessonWebsockets == null) {
    return;
}

lessonWebsockets.remove(login);
   }
 

public void sendMoreMessage(String[] userIds, String message) {
	for(String userId:userIds) {
		Session session = sessionPool.get(userId);
        if (session != null&&session.isOpen()) {
            try {
            	log.info("【websocket消息】 单点消息:"+message);
                session.getAsyncRemote().sendText(message);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
	}
    
}
 

@OnClose
public void close(Session session) {
    LOGGER.info("Closing socket {}", session.getId());
    if (cliCont != null) {
        cliCont.webSocketClosed();
    }
}
 

@Override
public void onError(Session session, Throwable thr) {
    if (onErrorConsumer != null) {
        onErrorConsumer.accept(thr);
    } else {
        LOGGER.log(Level.SEVERE, "No error handler defined. Received error was: ", thr);
    }
}
 

@org.junit.Test
public void testText() throws Exception {
    final byte[] payload = "payload".getBytes();
    final AtomicReference<Throwable> cause = new AtomicReference<>();
    final AtomicBoolean connected = new AtomicBoolean(false);
    final CompletableFuture<?> latch = new CompletableFuture<>();

    class TestEndPoint extends Endpoint {
        @Override
        public void onOpen(final Session session, EndpointConfig config) {
            connected.set(true);
            session.addMessageHandler(new MessageHandler.Whole<String>() {
                @Override
                public void onMessage(String message) {
                    session.getAsyncRemote().sendText(message);
                }
            });
        }
    }
    ServerWebSocketContainer builder = new ServerWebSocketContainer(TestClassIntrospector.INSTANCE, DefaultServer.getEventLoopSupplier(), Collections.EMPTY_LIST, false, false);
    builder.addEndpoint(ServerEndpointConfig.Builder.create(TestEndPoint.class, "/").configurator(new InstanceConfigurator(new TestEndPoint())).build());
    deployServlet(builder);

    WebSocketTestClient client = new WebSocketTestClient(new URI("ws://" + DefaultServer.getHostAddress("default") + ":" + DefaultServer.getHostPort("default") + "/"));
    client.connect();
    client.send(new TextWebSocketFrame(Unpooled.wrappedBuffer(payload)), new FrameChecker(TextWebSocketFrame.class, payload, latch));
    latch.get();
    Assert.assertNull(cause.get());
    client.destroy();
}
 

static void close() {
    for (Session session : sessions) {
        try {
            session.close();
        } catch (IOException ignored) {
        }
    }

}
 

@OnMessage
public void handleMessage(final byte[] message, Session session, boolean last) throws IOException {
    if (stream == null) {
        stream = session.getBasicRemote().getSendStream();
    }
    stream.write(message);
    stream.flush();
    if (last) {
        stream.close();
        stream = null;
    }
}
 

@OnMessage
public void echoTextMessage(Session session, String msg) {
    try {
        session.getBasicRemote().sendText(msg);
    } catch (IOException e) {
        try {
            session.close();
        } catch (IOException e1) {
            // Ignore
        }
    }
}
 

@Test
public void testConnectToServerEndpoint() throws Exception {
    Tomcat tomcat = getTomcatInstance();
    // No file system docBase required
    Context ctx = tomcat.addContext("", null);
    ctx.addApplicationListener(TesterEchoServer.Config.class.getName());
    Tomcat.addServlet(ctx, "default", new DefaultServlet());
    ctx.addServletMappingDecoded("/", "default");

    tomcat.start();

    WebSocketContainer wsContainer =
            ContainerProvider.getWebSocketContainer();
    // Set this artificially small to trigger
    // https://bz.apache.org/bugzilla/show_bug.cgi?id=57054
    wsContainer.setDefaultMaxBinaryMessageBufferSize(64);
    Session wsSession = wsContainer.connectToServer(
            TesterProgrammaticEndpoint.class,
            ClientEndpointConfig.Builder.create().build(),
            new URI("ws://" + getHostName() + ":" + getPort() +
                    TesterEchoServer.Config.PATH_ASYNC));
    CountDownLatch latch = new CountDownLatch(1);
    BasicText handler = new BasicText(latch);
    wsSession.addMessageHandler(handler);
    wsSession.getBasicRemote().sendText(MESSAGE_STRING_1);

    boolean latchResult = handler.getLatch().await(10, TimeUnit.SECONDS);

    Assert.assertTrue(latchResult);

    Queue<String> messages = handler.getMessages();
    Assert.assertEquals(1, messages.size());
    Assert.assertEquals(MESSAGE_STRING_1, messages.peek());

    ((WsWebSocketContainer) wsContainer).destroy();
}
 

@OnOpen
public void open(Session session) {
	System.out.println("open.");
	
	session.setMaxIdleTimeout(2 * TimeUnit.MILLISECONDS.convert(RCSharedConstants.MAX_ROUND_LENGTH_IN_NANOS, TimeUnit.NANOSECONDS));
	
	// Convert the session to a 'managed resource', so that it will automatically be disposed of once it expires.
	ResourceLifecycleUtil.getInstance().addNewSession(ServerWsClientUtil.convertSessionToManagedResource(session));
}
 

/**
 * A callback when created a new websocket connection.
 */
@OnOpen
public void onOpen(Session session) {
  logger.info("Connected ... " + session.getId());
  try {
    session.getBasicRemote().sendText("start");
  } catch (IOException e) {
    throw new RuntimeException(e);
  }
}
 

private void onCollaborativeInviteMessage(String sender, Session session, WebSocketMessage webSocketMessage) {

        String invitedUser = webSocketMessage.getString("remoteUser");
        JsonObject broadcastMessage = webSocketMessage.getJsonObject("broadcastMessage");
        String context = broadcastMessage.getString("context");
        String url = broadcastMessage.getString("url");

        CollaborativeRoom room = CollaborativeRoom.getByKeyName(webSocketMessage.getString("key"));

        if (webSocketSessionsManager.isAllowedToReachUser(sender, invitedUser)) {

            if (room.getMasterName().equals(sender) && room.findUserSession(invitedUser) == null) {
                // the master sent the invitation
                // the user is not already in the room
                if (!room.getPendingUsers().contains(invitedUser)) {
                    // the user is not yet in the pending list, add him.
                    room.addPendingUser(invitedUser);
                }

                String invite = "/invite " + url + "/room/" + room.getKey();

                JsonObjectBuilder b = Json.createObjectBuilder()
                        .add("type", CHAT_MESSAGE)
                        .add("remoteUser", sender)
                        .add("sender", sender)
                        .add("message", invite)
                        .add("context", context);
                WebSocketMessage message = new WebSocketMessage(b.build());
                webSocketSessionsManager.broadcast(invitedUser, message);
                broadcastNewContext(room);
            }


        }
    }
 

/**
 * Disconnects all sessions belonging to a given player identified by name. A disconnected session
 * is closed, the closure will trigger a notification on the client of the disconnected player.
 *
 * @param userName The name of the player whose sessions will be disconnected.
 * @param disconnectMessage Message that will be displayed to the disconnected player.
 * @return True if any sessions were disconnected, false if none (indicating player was no longer
 *     in chat).
 */
public boolean disconnectPlayerByName(final UserName userName, final String disconnectMessage) {
  final Set<Session> sessions =
      participants.values().stream()
          .filter(
              chatterSession ->
                  chatterSession.getChatParticipant().getUserName().equals(userName))
          .map(ChatterSession::getSession)
          .collect(Collectors.toSet());

  disconnectSession(sessions, disconnectMessage);
  return !sessions.isEmpty();
}
 

public static void closeOpeningSession(String deviceId) {
    Session openingSession = getOpeningSession(deviceId);
    if (openingSession != null) {
        try {
            openingSession.close();
        } catch (IOException e) {
            log.error("close session err, sessionId: {}", openingSession.getId());
        }
    }
}
 

@Override
public void onError(Session session, Throwable thr) {
  if (onErrorConsumer != null) {
    onErrorConsumer.accept(thr);
  } else {
    log.error("No error handler defined. Received error was: ", thr);
  }
}
 

/**
 * 有人进入房间
 *
 * @param session session
 */
@OnOpen
public void onOpen(Session session) {
    this.session = session;
    webSocketSet.add(this);
    addOnlineCount();
    LOGGER.info("有新用户加入!当前在线人数为:{}", getOnlineCount());
}
 

public void sendMessage(Session session, String message) {
    try {
        session.getBasicRemote().sendText(String.format("%s (From Server，Session ID=%s)", message, session.getId()));
    } catch (IOException e) {
        log.error("发送消息出错：{}", e.getMessage());
        e.printStackTrace();
    }
}
 

@Override
public Session connectToServer(Class<? extends Endpoint> endpointClass, ClientEndpointConfig cec, URI path)
		throws DeploymentException, IOException {

	throw new UnsupportedOperationException(
			"MockServerContainer does not support connectToServer(Class, ClientEndpointConfig, URI)");
}
 

public static void removeUserFromAllRoom(String callerLogin) {
    Set<Map.Entry<String, Room>> roomsEntries = new HashSet<>(DB.entrySet());
    for (Map.Entry<String, Room> entry : roomsEntries) {
        Session session = entry.getValue().getSessionForUserLogin(callerLogin);
        if (session != null) {
            entry.getValue().removeSession(session);
        }
    }
}
 

/**
 * 群发
 *
 * @param message
 *         消息内容
 * @param sessionSet
 *         客户端session列表
 * @throws IOException
 */
private static void broadcast(String message, Set<Session> sessionSet) {
    if (CollectionUtils.isEmpty(sessionSet)) {
        return;
    }
    // 多线程群发
    for (Session entry : sessionSet) {
        if (null != entry && entry.isOpen()) {
            sendMessage(message, entry);
        } else {
            sessionSet.remove(entry);
        }
    }
}
 

@Timed(name = "websocket_onClose_timer",
    reusable = true,
    tags = "label=websocket")
@Counted(name = "websocket_onClose_count",
    monotonic = true,
    reusable = true,
    tags = "label=websocket")
@Metered(name = "websocket_onClose_meter",
    reusable = true,
    tags = "label=websocket")
@OnClose
public void onClose(Session session, CloseReason r) {
    Log.log(Level.FINE, this, "A connection to the room has been closed with reason " + r);
}
 

/**
    * The scribe has submitted a report.
    */
   private static void submitReport(ObjectNode requestJSON, Session websocket) {
Long toolSessionId = Long
	.valueOf(websocket.getRequestParameterMap().get(AttributeNames.PARAM_TOOL_SESSION_ID).get(0));
String userName = websocket.getUserPrincipal().getName();
LearningWebsocketServer.getScribeService().submitReport(toolSessionId, userName, requestJSON);
   }