Java 类io.vertx.core.http.WebSocketFrame 实例源码

private void flushSlice(Buffer buffer, int from, int length, boolean isFinal) {
    // a slice is just a view on the same buffer: no need to copy something
    Buffer outBuf = buffer.slice(from, from + length);

    WebSocketFrame frame;
    if (!_firstFrameWritten) {
        if (_format == DataFormat.Binary) {
            frame = WebSocketFrame.binaryFrame(outBuf, isFinal);
        } else {
            frame = WebSocketFrame.textFrame(outBuf.toString("UTF-8"), isFinal);
        }
    } else {
        frame = WebSocketFrame.continuationFrame(outBuf, isFinal);
    }
    _socket.writeFrame(frame);
    _firstFrameWritten = true;
}

public void handleFrame(WebSocketFrame frame) {
    ByteBuffer buf = new VertxByteBuffer(frame.binaryData());
    boolean isFinal = frame.isFinal();
    boolean isText = frame.isText();

    try {
        if (isText || frame.isBinary()) {
            DataFormat format = isText ? DataFormat.Text : DataFormat.Binary;
            _msgReceiver.newMessage(format, buf, isFinal);
        } else if (frame.isContinuation()) {
            _msgReceiver.continueMessage(buf, isFinal);
        } else {
            throw new InvalidFrameException("Invalid frame type");
        }
    } catch (SqpException e) {
        _session.handleError(e);
    }
}

@Override
public boolean matches(Object argument) {
    if (!(argument instanceof WebSocketFrame)) {
        return false;
    }
    WebSocketFrame frame = (WebSocketFrame) argument;
    if (_isFinal != frame.isFinal()) {
        return false;
    }

    if (_frameType.equals(FrameType.Binary) && !frame.isBinary()) {
        return false;
    } else if (_frameType.equals(FrameType.Text) && !frame.isText()) {
        return false;
    } else if (_frameType.equals(FrameType.Continuation) && !frame.isContinuation()) {
        return false;
    }

    byte[] frameContent = frame.binaryData().getBytes();
    return Arrays.equals(_expectedContent, frameContent);
}

@Test
public void testSimpleString() throws InterruptedException {
    final String message = "xhello";

    getClient().websocket(8080, "localhost", SERVICE_REST_GET + "/testSimpleString", ws -> {
        long startTime = System.currentTimeMillis();
        ws.handler((data) -> {
            System.out.println("client testSimpleString:" + new String(data.getBytes()));
            assertNotNull(data.getString(0, data.length()));
            assertTrue(data.getString(0, data.length()).equals(message));
            ws.close();
            long endTime = System.currentTimeMillis();
            System.out.println("Total execution time testSimpleString: " + (endTime - startTime) + "ms");
            testComplete();
        });

        ws.writeFrame(WebSocketFrame.textFrame(message, true));
    });


    await();

}

@Test
public void testSendWebsocketContinuationFrames() {
  // Use raw websocket transport
  client.websocket("/echo/websocket", ws -> {

    int size = 65535;

    Buffer buffer1 = TestUtils.randomBuffer(size);
    Buffer buffer2 = TestUtils.randomBuffer(size);

    ws.writeFrame(io.vertx.core.http.WebSocketFrame.binaryFrame(buffer1, false));
    ws.writeFrame(io.vertx.core.http.WebSocketFrame.continuationFrame(buffer2, true));

    Buffer received = Buffer.buffer();

    ws.handler(buff -> {
      received.appendBuffer(buff);
      if (received.length() == size * 2) {
        testComplete();
      }
    });

  });

  await();
}

@Test
public void testTextFrameRawWebSocket() throws InterruptedException {
  String serverPath = "/textecho";
  setupSockJsServer(serverPath, this::echoRequest);

  String message = "hello";
  AtomicReference<String> receivedReply = new AtomicReference<>();
  WebSocket ws = setupRawWebsocketClient(serverPath);

  ws.handler(replyBuffer -> receivedReply.set(replyBuffer.toString()));

  ws.writeFrame(WebSocketFrame.textFrame(message, true));

  await(5, TimeUnit.SECONDS);

  assertEquals("Client reply should have matched request", message, receivedReply.get());
}

@Test
public void testTextFrameSockJs() throws InterruptedException {
  String serverPath = "/text-sockjs";
  setupSockJsServer(serverPath, this::echoRequest);

  List<Buffer> receivedMessages = new ArrayList<>();
  WebSocket openedWebSocket = setupSockJsClient(serverPath, receivedMessages);
  String messageToSend = "[\"testMessage\"]";
  openedWebSocket.writeFrame(WebSocketFrame.textFrame(messageToSend, true));

  await(5, TimeUnit.SECONDS);

  assertEquals("Client should have received 2 messages: the reply and the close.", 2, receivedMessages.size());
  Buffer expectedReply = Buffer.buffer("a" + messageToSend);
  assertEquals("Client reply should have matched request", expectedReply, receivedMessages.get(0));
  assertEquals("Final message should have been a close", SOCKJS_CLOSE_REPLY, receivedMessages.get(1));
}

private void handleFrame(final ServerWebSocket webSocket, final WebSocketFrame frame) {
    // Deserializing received message:
    final Object request = serializer.deserialize(frame.binaryData().getBytes());
    if (request instanceof MyJsonMessage) {
        System.out.println("Received message: " + ((MyJsonMessage) request).text);
    }

    // Sending a simple response message after 1 second:
    final MyJsonMessage response = new MyJsonMessage();
    response.id = idCounter.getAndIncrement();
    response.text = "Hello client ";
    vertx.setTimer(1000L, id -> webSocket.writeFinalBinaryFrame(Buffer.buffer(serializer.serialize(response))));
}

private void handleJsonFrame(final ServerWebSocket webSocket, final WebSocketFrame frame) {
    final byte[] packet = frame.binaryData().getBytes();
    final long start = System.nanoTime();
    final Object deserialized = jsonSerializer.deserialize(packet);
    final long time = System.nanoTime() - start;

    final com.github.czyzby.shared.json.ServerResponse response = new com.github.czyzby.shared.json.ServerResponse();
    response.message = "Packet had " + packet.length + " bytes. Class: " + deserialized.getClass().getSimpleName()
            + ", took " + time + " nanos to deserialize.";
    System.out.println(response.message);
    final byte[] serialized = jsonSerializer.serialize(response);
    webSocket.writeFinalBinaryFrame(Buffer.buffer(serialized));
}

private void handleSerializationFrame(final ServerWebSocket webSocket, final WebSocketFrame frame) {
    final byte[] packet = frame.binaryData().getBytes();
    final long start = System.nanoTime();
    final Object deserialized = serializer.deserialize(packet);
    final long time = System.nanoTime() - start;

    final ServerResponse response = new ServerResponse("Packet had " + packet.length + " bytes. Class: "
            + deserialized.getClass().getSimpleName() + ", took " + time + " nanos to deserialize.");
    System.out.println(response.getMessage());
    final byte[] serialized = serializer.serialize(response);
    webSocket.writeFinalBinaryFrame(Buffer.buffer(serialized));
}

/**
 * Writing multiple continuation frames from the client side should result in a single message on the server side
 * after the frames are re-combined
 */
@Test
public void testCombineBinaryContinuationFramesRawWebSocket() throws InterruptedException {
  String serverPath = "/combine";

  AtomicReference<Buffer> serverReceivedMessage = new AtomicReference<>();
  setupSockJsServer(serverPath, (sock, requestBuffer) -> {
    serverReceivedMessage.set(requestBuffer);
    sock.write(Buffer.buffer("reply"));
    sock.close();
  });


  Buffer largeMessage = Buffer.buffer(TestUtils.randomAlphaString(30));
  WebSocketFrame frame1 = WebSocketFrame.binaryFrame(largeMessage.slice(0, 10), false);
  WebSocketFrame frame2 = WebSocketFrame.continuationFrame(largeMessage.slice(10, 20), false);
  WebSocketFrame frame3 = WebSocketFrame.continuationFrame(largeMessage.slice(20, largeMessage.length()), true);

  WebSocket ws = setupRawWebsocketClient(serverPath);
  ws.writeFrame(frame1);
  ws.writeFrame(frame2);
  ws.writeFrame(frame3);

  await(5, TimeUnit.SECONDS);

  assertEquals("Server did not combine continuation frames correctly", largeMessage, serverReceivedMessage.get());
}

@Test
public void testSplitLargeReplyRawWebSocket() throws InterruptedException {
  String serverPath = "/split";

  String largeReply = TestUtils.randomAlphaString(65536 * 5);
  Buffer largeReplyBuffer = Buffer.buffer(largeReply);

  setupSockJsServer(serverPath, (sock, requestBuffer) -> {
    sock.write(largeReplyBuffer);
    sock.close();
  });

  Buffer totalReplyBuffer = Buffer.buffer(largeReplyBuffer.length());
  AtomicInteger receivedReplies = new AtomicInteger(0);
  WebSocket ws = setupRawWebsocketClient(serverPath);
  ws.handler(replyBuffer -> {
            totalReplyBuffer.appendBuffer(replyBuffer);
            receivedReplies.incrementAndGet();

          });

  ws.writeFrame(WebSocketFrame.binaryFrame(Buffer.buffer("hello"), true));

  await(5, TimeUnit.SECONDS);

  int receivedReplyCount = receivedReplies.get();
  assertEquals("Combined reply on client should equal message from server", largeReplyBuffer, totalReplyBuffer);
  assertTrue("Should have received > 1 reply frame, actually received " + receivedReplyCount, receivedReplyCount > 1);
}

@Test
public void testCombineTextFrameSockJs() throws InterruptedException {
  String serverPath = "/text-combine-sockjs";
  setupSockJsServer(serverPath, this::echoRequest);

  List<Buffer> receivedMessages = new ArrayList<>();
  WebSocket openedWebSocket = setupSockJsClient(serverPath, receivedMessages);

  Buffer largeMessage = Buffer.buffer("[\"" + TestUtils.randomAlphaString(30) + "\"]");
  WebSocketFrame frame1 = new WebSocketFrameImpl(FrameType.TEXT, largeMessage.slice(0, 10).getByteBuf(), false);
  WebSocketFrame frame2 = WebSocketFrame.continuationFrame(largeMessage.slice(10, 20), false);
  WebSocketFrame frame3 = WebSocketFrame.continuationFrame(largeMessage.slice(20, largeMessage.length()), true);

  log.debug("Client sending " + frame1.textData());
  openedWebSocket.writeFrame(frame1);
  log.debug("Client sending " + frame2.textData());
  openedWebSocket.writeFrame(frame2);
  log.debug("Client sending " + frame3.textData());
  openedWebSocket.writeFrame(frame3);

  await(5, TimeUnit.SECONDS);

  assertEquals("Client should have received 2 messages: the reply and the close.", 2, receivedMessages.size());
  Buffer expectedReply = Buffer.buffer("a" + largeMessage.toString());
  assertEquals("Client reply should have matched request", expectedReply, receivedMessages.get(0));
  assertEquals("Final message should have been a close", SOCKJS_CLOSE_REPLY, receivedMessages.get(1));
}

@Test
public void testSplitLargeReplySockJs() throws InterruptedException {
  String serverPath = "/large-reply-sockjs";

  String largeMessage = TestUtils.randomAlphaString(65536 * 2);
  Buffer largeReplyBuffer = Buffer.buffer(largeMessage);

  setupSockJsServer(serverPath, (sock, requestBuffer) -> {
    sock.write(largeReplyBuffer);
    sock.close();
  });

  List<Buffer> receivedMessages = new ArrayList<>();
  WebSocket openedWebSocket = setupSockJsClient(serverPath, receivedMessages);

  String messageToSend = "[\"hello\"]";
  openedWebSocket.writeFrame(WebSocketFrame.textFrame(messageToSend, true));

  await(5, TimeUnit.SECONDS);

  int receivedReplyCount = receivedMessages.size();
  assertTrue("Should have received > 2 reply frame, actually received " + receivedReplyCount, receivedReplyCount > 2);

  Buffer expectedReplyBuffer = Buffer.buffer("a[\"").appendBuffer(largeReplyBuffer).appendBuffer(Buffer.buffer("\"]"));
  Buffer clientReplyBuffer = combineReplies(receivedMessages.subList(0, receivedMessages.size() - 1));
  assertEquals(String.format("Combined reply on client (length %s) should equal message from server (%s)",
          clientReplyBuffer.length(), expectedReplyBuffer.length()),
          expectedReplyBuffer, clientReplyBuffer);

  Buffer finalMessage = receivedMessages.get(receivedMessages.size() - 1);
  assertEquals("Final message should have been a close", SOCKJS_CLOSE_REPLY, finalMessage);
}

private Handler<WebSocketFrame> frameHandler()
{
    return frame -> {
        if (frame.isBinary()) {
            throw new IllegalArgumentException("Binary messages are not supported by this server");
        }

        // TODO this doesn't handle multi-frame messages
        String message = frame.textData();

        logger.debug("[SERVER] Received WAMP message {}", message);
        passMessageToWampListener(message, objectMapper, connectionListener);
    };
}

public static WebSocketFrame binaryFrameEq(boolean isFinal, byte[] content) {
    return argThat(new WebSocketFrameMatcher(isFinal, WebSocketFrameMatcher.FrameType.Binary, content));
}

public static WebSocketFrame textFrameEq(boolean isFinal, String content) {
    byte[] bytes = content.getBytes(StandardCharsets.UTF_8);
    return argThat(new WebSocketFrameMatcher(isFinal, WebSocketFrameMatcher.FrameType.Text, bytes));
}

public static WebSocketFrame continuationFrameEq(boolean isFinal, byte[] content) {
    return argThat(new WebSocketFrameMatcher(isFinal, WebSocketFrameMatcher.FrameType.Continuation, content));
}

@Test
/*
    Constructs a message with size == 2*MAX_WEBSOCKET_FRAME_SIZE. The message is then sent via
    eventBus bridge. The test then reads the message via WebSocket and makes sure that the message
    is delivered in three WebSocketFrames.
    Regression for #35
 */
public void testSendingAMessageBiggerThanSocketFrameSize() {
  AtomicReference<Throwable> error = new AtomicReference<>();
  List<WebSocketFrame> wsBuffers = new ArrayList<>();
  List<Buffer> stompBuffers = new ArrayList<>();

  AtomicReference<WebSocket> socket = new AtomicReference<>();
  AtomicReference<StompClientConnection> client = new AtomicReference<>();

  clients.add(StompClient.create(vertx).connect(61613, "localhost", connection -> {
    connection.result().subscribe("bigData", h-> {}, r -> {
      client.set(connection.result());

    });
    connection.result().receivedFrameHandler(stompFrame -> {
      if(stompFrame.toBuffer().toString().startsWith("MESSAGE")) {
        stompBuffers.add(stompFrame.toBuffer());
      }
    });
  }));

  vertx.createHttpClient().websocket(8080, "localhost", "/stomp", MultiMap.caseInsensitiveMultiMap().add
    ("Sec-WebSocket-Protocol", "v10.stomp, v11.stomp, v12.stomp"), ws -> {
      ws.exceptionHandler(error::set)
        .handler(buffer -> {
          if (buffer.toString().startsWith("CONNECTED")) {
            ws.write(
              new Frame(Frame.Command.SUBSCRIBE, Headers.create("id", "myId", "destination", "bigData"), null)
                .toBuffer());
            return;
          }
          // Start collecting the frames once we see the first real payload message
          if (buffer.toString().startsWith("MESSAGE")) {
            ws.frameHandler(wsBuffers::add);
          }
        })
        .write(new Frame(Frame.Command.CONNECT, Headers.create("accept-version", "1.2,1.1,1.0",
          "heart-beat", "10000,10000"), null).toBuffer());
    socket.set(ws);
  });

  // Create content that is slightly bigger than the size of a single web socket frame
  String bufferContent = StringUtils.repeat("*",  2 * MAX_WEBSOCKET_FRAME_SIZE);

  await().atMost(10, TimeUnit.SECONDS).until(() -> client.get() != null);
  await().atMost(10, TimeUnit.SECONDS).until(() -> socket.get() != null);
  vertx.eventBus().publish("bigData",bufferContent);

  await().atMost(10, TimeUnit.SECONDS).until(() -> error.get() == null && stompBuffers.size() == 1);
  await().atMost(10, TimeUnit.SECONDS).until(() -> error.get() == null && wsBuffers.size() == 3);

  // STOMP message has 2048 bytes of payload + headers => 2167 bytes
  assertEquals(2167, stompBuffers.get(0).getBytes().length);

  // We expect two complete frames + 1 with 116 bytes
  assertEquals(MAX_WEBSOCKET_FRAME_SIZE, wsBuffers.get(0).binaryData().getBytes().length);
  assertEquals(MAX_WEBSOCKET_FRAME_SIZE, wsBuffers.get(1).binaryData().getBytes().length);
  assertEquals(116, wsBuffers.get(2).binaryData().getBytes().length);
  socket.get().close();
}

private static void handleStringFrame(final ServerWebSocket webSocket, final WebSocketFrame frame) {
    final String response = "Packet had " + frame.binaryData().length()
            + " bytes. Cannot deserialize packet class.";
    System.out.println(response);
    webSocket.writeFinalTextFrame(response);
}

public EventToJsonAction(Observable<Event> eventObservable, Func1<String, WebSocketFrame> frameFactory) {
    this.frameFactory = frameFactory;
    this.eventObservable = eventObservable;
}