Java 类javax.websocket.RemoteEndpoint.Basic 实例源码

synchronized LSPProcess createProcess(String wsKey, String lang, RemoteEndpoint.Basic remoteEndpoint, String ownerSessionId) throws LSPException {

        String procKey = processKey(wsKey, lang);
        String rpcType = langContexts.get(lang).getRpcType();
        String wsKeyElem[] = wsKey.split(WS_KEY_DELIMITER,3);

        disconnect(lang, ownerSessionId);
        LSPProcess lspProcess = new LSPProcess(wsKeyElem, lang, langContexts.get(lang).getProcessBuilder(wsKeyElem), remoteEndpoint, ownerSessionId);
        switch(rpcType) {
        case ENV_IPC_SOCKET:
            socketEnv(lspProcess, LangServerCtx.LangPrefix(lang));
            break;
        case ENV_IPC_PIPES:
            pipeEnv(lspProcess, LangServerCtx.LangPrefix(lang));
            break;
        case ENV_IPC_CLIENT:
            clientSocketEnv(lspProcess, LangServerCtx.LangPrefix(lang));
            break;
        default:
            streamEnv(lspProcess);
        }
        lspProcesses.put(procKey, lspProcess);
        return lspProcess;
    }

@OnOpen
public void open(@PathParam("gametype") String gameID, Session session) throws IOException {
    type = GameType.getGameType(gameID);
    if(type == null) {
        session.close(new CloseReason(CloseReason.CloseCodes.UNEXPECTED_CONDITION, "Invalid game type"));
        return;
    }

    Basic sender = session.getBasicRemote();

    viewer = data -> {
            synchronized(session) {
                if(session.isOpen())
                    try { sender.sendBinary(data); } catch (IOException e) {}
            }
    };

    DisplayHandler.addGlobalViewer(viewer);
}

@Test
public void onError() throws IOException, EncodeException {
    final Session session = mock(Session.class);
    when(session.getId()).thenReturn("sessionId");
    final Throwable cause = new Throwable();
    when(this.errorEvent.select(OnError.Literal.onError())).thenReturn(this.errorEvent);
    final Basic basic = mock(Basic.class);
    when(session.getBasicRemote()).thenReturn(basic);

    this.endpoint.onError(session, cause);

    verify(session).getId();
    verify(session).getUserPrincipal();
    verify(this.log).warn(eq("WebSocket error. [id={},principle={},errorId={}]"), eq("sessionId"), isNull(), anyString(), eq(cause));
    verify(this.errorEvent).select(OnError.Literal.onError());
    verify(this.errorEvent).fire(cause);
    verify(session).getBasicRemote();
    verify(basic).sendObject(any(Frame.class));
    verify(session).close(any(CloseReason.class));
    verifyNoMoreInteractions(session, basic);
}

@Test
public void fromBroker() throws IOException, EncodeException {
    final Message msg = mock(Message.class);
    when(msg.sessionId()).thenReturn("sessionId");
    final Frame frame = mock(Frame.class);
    when(msg.frame()).thenReturn(frame);
    when(frame.command()).thenReturn(Command.MESSAGE);
    final Session session = Mockito.mock(Session.class);
    getSessionMap().put("sessionId", session);
    getPrincipalSessionMap().put(NULL_PRINCIPLE, new HashSet<>(singleton(session)));
    final Basic basic = mock(Basic.class);
    when(session.getBasicRemote()).thenReturn(basic);

    this.registry.fromBroker(msg);

    verify(msg).sessionId();
    verify(msg).frame();
    verify(frame, times(2)).command();
    verify(this.log).debug("Sending message to client. [sessionId={},command={}]", "sessionId", Command.MESSAGE);
    verify(session).getBasicRemote();
    verify(basic).sendObject(frame);
    verifyNoMoreInteractions(msg, frame, session, basic);
}

@Test
public void fromBroker_ioe() throws IOException, EncodeException {
    final Message msg = mock(Message.class);
    when(msg.sessionId()).thenReturn("sessionId");
    final Frame frame = mock(Frame.class);
    when(msg.frame()).thenReturn(frame);
    when(frame.command()).thenReturn(Command.MESSAGE);
    final Session session = Mockito.mock(Session.class);
    getSessionMap().put("sessionId", session);
    getPrincipalSessionMap().put(NULL_PRINCIPLE, new HashSet<>(singleton(session)));
    final Basic basic = mock(Basic.class);
    when(session.getBasicRemote()).thenReturn(basic);
    final IOException ioe = new IOException();
    doThrow(ioe).when(basic).sendObject(frame);

    this.registry.fromBroker(msg);

    verify(msg).sessionId();
    verify(msg).frame();
    verify(frame, times(3)).command();
    verify(this.log).debug("Sending message to client. [sessionId={},command={}]", "sessionId", Command.MESSAGE);
    verify(session).getBasicRemote();
    verify(basic).sendObject(frame);
    verify(this.log).error("Unable to send message! [sessionid={},command={}]", "sessionId", Command.MESSAGE, ioe);
    verifyNoMoreInteractions(msg, frame, session, basic);
}

public long copyInputToOutput() throws IOException {
    Basic basicRemote = session.getBasicRemote();
    OutputStream outputStream = basicRemote.getSendStream();

    try {
        // slurp the input stream data and send directly to the output stream
        byte[] buf = new byte[4096];
        long totalBytesCopied = 0L;
        while (true) {
            int numRead = inputStream.read(buf);
            if (numRead == -1) {
                break;
            }
            outputStream.write(buf, 0, numRead);
            totalBytesCopied += numRead;
        }
        return totalBytesCopied;
    } finally {
        try {
            outputStream.close();
        } finally {
            inputStream.close();
        }
    }
}

LSPProcess(String wsKeyElem[], String lang, ProcessBuilder pb, Basic remoteEndpoint, String ownerSessionId) {
    this.pb = pb;
    this.remoteEndpoint = remoteEndpoint; 
    this.projPathElem = "/" + String.join("/", Arrays.copyOfRange(wsKeyElem,1,wsKeyElem.length));
    this.lang = lang;
    this.ownerSessionId = ownerSessionId;
}

@OnMessage
public void onMessage(Session session, String msg) throws IOException {

    if (started) {
        return;
    }
    synchronized (this) {
        if (started) {
            return;
        } else {
            started = true;
        }
    }

    System.out.println("Received " + msg + ", now sending data");

    session.getUserProperties().put(
            "org.apache.tomcat.websocket.BLOCKING_SEND_TIMEOUT",
            Long.valueOf(SEND_TIME_OUT_MILLIS));

    Basic remote = session.getBasicRemote();
    remote.setBatchingAllowed(true);

    for (int i = 0; i < MESSAGE_COUNT; i++) {
        remote.sendText(MESSAGE);
        if (i % (MESSAGE_COUNT * 0.4) == 0) {
            remote.setBatchingAllowed(false);
            remote.setBatchingAllowed(true);
        }
    }

    // Flushing should happen automatically on session close
    session.close();
}

private Session getSession() {
    final Basic basicRemote = mock(Basic.class);

    final Session session = mock(Session.class);
    when(session.getBasicRemote()).thenReturn(basicRemote);
    return session;
}

@OnMessage
public void onMessage(Session session, String msg) throws IOException {

    if (started) {
        return;
    }
    synchronized (this) {
        if (started) {
            return;
        } else {
            started = true;
        }
    }

    System.out.println("Received " + msg + ", now sending data");

    session.getUserProperties().put(
            "org.apache.tomcat.websocket.BLOCKING_SEND_TIMEOUT",
            Long.valueOf(SEND_TIME_OUT_MILLIS));

    Basic remote = session.getBasicRemote();
    remote.setBatchingAllowed(true);

    for (int i = 0; i < MESSAGE_COUNT; i++) {
        remote.sendText(MESSAGE);
        if (i % (MESSAGE_COUNT * 0.4) == 0) {
            remote.setBatchingAllowed(false);
            remote.setBatchingAllowed(true);
        }
    }

    // Flushing should happen automatically on session close
    session.close();
}

@Test
   public void test() throws Exception {
Session session = webSocketContainer.connectToServer(this, new URI(
    "ws://localhost:8080/purifinityserver/socket/server"));
try {
    Basic basic = session.getBasicRemote();
    basic.sendText("getStatus");
} finally {
    session.close(new CloseReason(CloseCodes.GOING_AWAY,
        "We are done..."));
}
Thread.sleep(5000);
   }

@OnMessage
public void onMessage(Session session, String msg) throws IOException {

    if (started) {
        return;
    }
    synchronized (this) {
        if (started) {
            return;
        } else {
            started = true;
        }
    }

    System.out.println("Received " + msg + ", now sending data");

    session.getUserProperties().put(
            "org.apache.tomcat.websocket.BLOCKING_SEND_TIMEOUT",
            Long.valueOf(SEND_TIME_OUT_MILLIS));

    Basic remote = session.getBasicRemote();
    remote.setBatchingAllowed(true);

    for (int i = 0; i < MESSAGE_COUNT; i++) {
        remote.sendText(MESSAGE);
    }

    // Ensure remaining messages are flushed
    remote.setBatchingAllowed(false);
}

@OnMessage
public void onMessage(Session session, String msg) throws IOException {

    if (started) {
        return;
    }
    synchronized (this) {
        if (started) {
            return;
        } else {
            started = true;
        }
    }

    System.out.println("Received " + msg + ", now sending data");

    session.getUserProperties().put(
            "org.apache.tomcat.websocket.BLOCKING_SEND_TIMEOUT",
            Long.valueOf(SEND_TIME_OUT_MILLIS));

    Basic remote = session.getBasicRemote();
    remote.setBatchingAllowed(true);

    for (int i = 0; i < MESSAGE_COUNT; i++) {
        remote.sendText(MESSAGE);
    }

    // Ensure remaining messages are flushed
    remote.setBatchingAllowed(false);
}

@OnMessage
public void onMessage(Session session, String msg) throws IOException {

    if (started) {
        return;
    }
    synchronized (this) {
        if (started) {
            return;
        } else {
            started = true;
        }
    }

    System.out.println("Received " + msg + ", now sending data");

    session.getUserProperties().put(
            "org.apache.tomcat.websocket.BLOCKING_SEND_TIMEOUT",
            Long.valueOf(SEND_TIME_OUT_MILLIS));

    Basic remote = session.getBasicRemote();
    remote.setBatchingAllowed(true);

    for (int i = 0; i < MESSAGE_COUNT; i++) {
        remote.sendText(MESSAGE);
    }

    // Ensure remaining messages are flushed
    remote.setBatchingAllowed(false);
}

OutputStreamHandler(RemoteEndpoint.Basic remoteEndpointBasic, InputStream out) {
    this.remote = remoteEndpointBasic;
    this.out = out;
}

public void sendMessage(Message message) throws IOException{
    Basic basic= session.getBasicRemote();
    basic.sendText(message.getString());
}

@Override
public Basic getBasicRemote() {
    Basic basic= new MockBasic();
    return basic;
}

public void sendMessage(Message message) throws IOException{
    Basic basic= session.getBasicRemote(); // MockWebsocketSession returns a MockBasic
    basic.sendText(message.getString()); // MockBasic sendText does nothing
}

public void sendTextSync(Session session, String text) throws IOException {
    Basic basicRemote = session.getBasicRemote();
    basicRemote.sendText(text);
}

public Basic getBasicRemote() {
    // TODO Auto-generated method stub
    return null;
}