java类io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame的实例源码

@Override
public void channelRead0(ChannelHandlerContext ctx, Object msg) {
    Channel ch = ctx.channel();
    if (!handshaker.isHandshakeComplete()) {
        handshaker.finishHandshake(ch, (FullHttpResponse) msg);
        System.out.println("WebSocket Client UID:[" + this.uid + "] handshaker connected!");
        handshakeFuture.setSuccess();
        return;
    }

    if (msg instanceof BinaryWebSocketFrame) {
        try {
            Object obj = protobufDecoder.decode(((BinaryWebSocketFrame) msg).content());
            resQueue.add((Response.HeshResMessage)obj);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

@Override
protected void encode(ChannelHandlerContext ctx, Proto proto, List<Object> list) throws Exception {
    ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
    if (proto.getBody() != null) {
        byteBuf.writeInt(Proto.HEADER_LENGTH + proto.getBody().length);
        byteBuf.writeShort(Proto.HEADER_LENGTH);
        byteBuf.writeShort(Proto.VERSION);
        byteBuf.writeInt(proto.getOperation());
        byteBuf.writeInt(proto.getSeqId());
        byteBuf.writeBytes(proto.getBody());
    } else {
        byteBuf.writeInt(Proto.HEADER_LENGTH);
        byteBuf.writeShort(Proto.HEADER_LENGTH);
        byteBuf.writeShort(Proto.VERSION);
        byteBuf.writeInt(proto.getOperation());
        byteBuf.writeInt(proto.getSeqId());
    }

    list.add(new BinaryWebSocketFrame(byteBuf));

    logger.debug("encode: {}", proto);
}

@Test
public void testVersion() throws Exception {
    try {
        String uuid = UUID.randomUUID().toString();
        VersionRequest request = new VersionRequest();
        request.setRequestId(uuid);
        ch.writeAndFlush(new BinaryWebSocketFrame(Unpooled.wrappedBuffer(JsonSerializer.getObjectMapper()
                .writeValueAsBytes(request))));
        // Confirm receipt of all data sent to this point
        List<byte[]> response = handler.getResponses();
        while (response.size() == 0 && handler.isConnected()) {
            LOG.info("Waiting for web socket response");
            sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
            response = handler.getResponses();
        }
        assertEquals(1, response.size());
        VersionResponse version = JsonSerializer.getObjectMapper()
                .readValue(response.get(0), VersionResponse.class);
        assertEquals(VersionResponse.VERSION, version.getVersion());
        assertEquals(uuid, version.getRequestId());
    } finally {
        ch.close().sync();
        s.shutdown();
        group.shutdownGracefully();
    }
}

@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
   protected void encode(ChannelHandlerContext ctx, Packet packet, List out) throws Exception {
       ByteBuf buf = ctx.alloc().buffer().order(ByteOrder.LITTLE_ENDIAN);
       int packetId = PacketRegistry.SERVER2CLIENT.getPacketId(packet.getClass());
       if (packetId == -1) {
           throw new IllegalArgumentException("Provided packet is not registered as a clientbound packet!");
       }

       buf.writeByte(packetId);
       packet.writeData(buf);
       new BinaryWebSocketFrame(buf);
       out.add(new BinaryWebSocketFrame(buf));

       Log.logDebug("Sent packet ID " + packetId + " (" + packet.getClass().getSimpleName() + ") to " + ctx.channel().remoteAddress());
   }

private void handleFrame(Channel channel, WebSocketFrame frame, WebSocketUpgradeHandler handler, NettyWebSocket webSocket) throws Exception {
    if (frame instanceof CloseWebSocketFrame) {
        Channels.setDiscard(channel);
        CloseWebSocketFrame closeFrame = (CloseWebSocketFrame) frame;
        webSocket.onClose(closeFrame.statusCode(), closeFrame.reasonText());
    } else {
        ByteBuf buf = frame.content();
        if (buf != null && buf.readableBytes() > 0) {
            HttpResponseBodyPart part = config.getResponseBodyPartFactory().newResponseBodyPart(buf, frame.isFinalFragment());
            handler.onBodyPartReceived(part);

            if (frame instanceof BinaryWebSocketFrame) {
                webSocket.onBinaryFragment(part);
            } else if (frame instanceof TextWebSocketFrame) {
                webSocket.onTextFragment(part);
            } else if (frame instanceof PingWebSocketFrame) {
                webSocket.onPing(part);
            } else if (frame instanceof PongWebSocketFrame) {
                webSocket.onPong(part);
            }
        }
    }
}

private Message decodeWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
    // Check for closing frame
    if (frame instanceof CloseWebSocketFrame) {
        handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
        return null;
    }

    if (frame instanceof PingWebSocketFrame) {
        ctx.write(new PongWebSocketFrame(frame.content().retain()));
        return null;
    }

    if (frame instanceof TextWebSocketFrame) {
        TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
        return parseMessage(textFrame.content());
    }

    if (frame instanceof BinaryWebSocketFrame) {
        BinaryWebSocketFrame binFrame = (BinaryWebSocketFrame) frame;
        return parseMessage(binFrame.content());
    }

    log.warn("Message format error: " + frame); 
    return null;
}

private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {

        // Check for closing frame
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            return;
        }
        if (frame instanceof PingWebSocketFrame) {
            ctx.write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        if (frame instanceof TextWebSocketFrame) {
            // Echo the frame
            ctx.write(frame.retain());
            return;
        }
        if (frame instanceof BinaryWebSocketFrame) {
            // Echo the frame
            ctx.write(frame.retain());
            return;
        }
    }

@Override
public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
    System.out.println("client channelRead0 "+ctx);
    Channel ch = ctx.channel();
    if (!handshaker.isHandshakeComplete()) {
        handshaker.finishHandshake(ch, (FullHttpResponse) msg);
        System.out.println("WebSocket Client connected!");
        handshakeFuture.setSuccess();
    }

    if(msg instanceof WebSocketFrame){
        WebSocketFrame frame = (WebSocketFrame)msg;
        if(frame instanceof BinaryWebSocketFrame){
            handleWebSocketFrame(ctx,  frame);
        }
        return;
    }
    return;
}

@Override
public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
    System.out.println("client channelRead0 "+ctx);
    Channel ch = ctx.channel();
    if (!handshaker.isHandshakeComplete()) {
        handshaker.finishHandshake(ch, (FullHttpResponse) msg);
        System.out.println("WebSocket Client connected!");
        handshakeFuture.setSuccess();
    }

    if(msg instanceof WebSocketFrame){
        WebSocketFrame frame = (WebSocketFrame)msg;
        if(frame instanceof BinaryWebSocketFrame){
            handleWebSocketFrame(ctx,  frame);
        }
        return;
    }
    sendRealTimeMessageTest(ctx);
    return;
}

@Override
protected void channelRead0(ChannelHandlerContext context, Object message) throws Exception {
    final Channel channel = context.channel();
    if (!handshaker.isHandshakeComplete()) {
        handshaker.finishHandshake(channel, (FullHttpResponse) message);
        channel.pipeline().addBefore(HANDLER_NAME, "websocket-frame-aggregator", new WebSocketFrameAggregator(64 * 1024));
        subscriber.onStart();
        return;
    }

    if (message instanceof FullHttpResponse) {
        final FullHttpResponse response = (FullHttpResponse) message;
        throw new IllegalStateException(
                "Unexpected FullHttpResponse (getStatus=" + response.getStatus() +
                        ", content=" + response.content().toString(CharsetUtil.UTF_8) + ')');
    }

    final WebSocketFrame frame = (WebSocketFrame) message;
    if (frame instanceof PingWebSocketFrame) {
        context.writeAndFlush(new PongWebSocketFrame(((PingWebSocketFrame)frame).retain().content()));
    } else if (frame instanceof BinaryWebSocketFrame) {
        final ByteBufInputStream input = new ByteBufInputStream(((BinaryWebSocketFrame)message).content());
        final Envelope envelope = Envelope.ADAPTER.decode(input);
        subscriber.onNext(envelope);
    }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {

        // Check for closing frame
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            return;
        }
        if (frame instanceof PingWebSocketFrame) {
            ctx.write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        if (frame instanceof TextWebSocketFrame) {
            // Echo the frame
            ctx.write(frame.retain());
            return;
        }
        if (frame instanceof BinaryWebSocketFrame) {
            // Echo the frame
            ctx.write(frame.retain());
            return;
        }
    }

private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
    if (logger.isLoggable(Level.FINE)) {
        logger.fine(String.format(
                "Channel %s received %s", ctx.channel().hashCode(), StringUtil.simpleClassName(frame)));
    }

    if (frame instanceof CloseWebSocketFrame) {
        handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame);
    } else if (frame instanceof PingWebSocketFrame) {
        ctx.write(new PongWebSocketFrame(frame.isFinalFragment(), frame.rsv(), frame.content()), ctx.voidPromise());
    } else if (frame instanceof TextWebSocketFrame) {
        ctx.write(frame, ctx.voidPromise());
    } else if (frame instanceof BinaryWebSocketFrame) {
        ctx.write(frame, ctx.voidPromise());
    } else if (frame instanceof ContinuationWebSocketFrame) {
        ctx.write(frame, ctx.voidPromise());
    } else if (frame instanceof PongWebSocketFrame) {
        frame.release();
        // Ignore
    } else {
        throw new UnsupportedOperationException(String.format("%s frame types not supported", frame.getClass()
                .getName()));
    }
}

private ReactiveSocketWebSocketClient(WebSocketConnection wsConn) {
    this.reactiveSocket = ReactiveSocket.createRequestor();
    connect = this.reactiveSocket.connect(
            new DuplexConnection() {
                @Override
                public Publisher<Frame> getInput() {
                    return toPublisher(wsConn.getInput().map(frame -> {
                        return Frame.from(frame.content().nioBuffer());
                    }));
                }

                @Override
                public Publisher<Void> addOutput(Publisher<Frame> o) {
                    // had to use writeAndFlushOnEach instead of write for frames to get through
                    // TODO determine if that's expected or not
                    Publisher<Void> p = toPublisher(wsConn.writeAndFlushOnEach(toObservable(o)
                            .map(frame -> new BinaryWebSocketFrame(Unpooled.wrappedBuffer(frame.getByteBuffer())))
                    ));
                    return p;
                }
            });
}

/**
   * Use this method as the RxNetty HttpServer WebSocket handler.
   * 
   * @param ws
   * @return
   */
  public Observable<Void> acceptWebsocket(WebSocketConnection ws) {
    return toObservable(reactiveSocket.connect(new DuplexConnection() {
    @Override
    public Publisher<Frame> getInput() {
        return toPublisher(ws.getInput().map(frame -> {
            // TODO is this copying bytes?
            try {
                return Frame.from(frame.content().nioBuffer());
            } catch (Exception e) {
                e.printStackTrace();
                throw new RuntimeException(e);
            }
        }));
    }

    @Override
    public Publisher<Void> addOutput(Publisher<Frame> o) {
        // had to use writeAndFlushOnEach instead of write for frames to reliably get through
        // TODO determine if that's expected or not
        return toPublisher(ws.writeAndFlushOnEach(toObservable(o).map(frame -> {
            return new BinaryWebSocketFrame(Unpooled.wrappedBuffer(frame.getByteBuffer()));
        })));
    }
}));
  }

@Override
public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
  Channel ch = ctx.channel();

  if (!handshaker.isHandshakeComplete()) {
    handshaker.finishHandshake(ch, (FullHttpResponse) msg);
    handshakeFuture.setSuccess();
    return;
  }

  if (!(msg instanceof BinaryWebSocketFrame)) {
    ch.close();
    log.warn("Received {}, closing", msg);
    return;
  }

  byte[] b = extractBytes(msg);

  ctx.fireChannelRead(b);
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
        throws UnknownWebSocketFrameTypeException, ServerConnectorException {
    if (!(msg instanceof WebSocketFrame)) {
        logger.error("Expecting WebSocketFrame. Unknown type.");
        throw new UnknownWebSocketFrameTypeException("Expecting WebSocketFrame. Unknown type.");
    }
    if (msg instanceof TextWebSocketFrame) {
        notifyTextMessage((TextWebSocketFrame) msg);
    } else if (msg instanceof BinaryWebSocketFrame) {
        notifyBinaryMessage((BinaryWebSocketFrame) msg);
    } else if (msg instanceof CloseWebSocketFrame) {
        notifyCloseMessage((CloseWebSocketFrame) msg);
    } else if (msg instanceof PingWebSocketFrame) {
        notifyPingMessage((PingWebSocketFrame) msg);
    } else if (msg instanceof PongWebSocketFrame) {
        notifyPongMessage((PongWebSocketFrame) msg);
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) throws Exception {
    if (frame instanceof TextWebSocketFrame) {
        // Echos the same text
        String text = ((TextWebSocketFrame) frame).text();
        if (PING.equals(text)) {
            ctx.writeAndFlush(new PingWebSocketFrame(Unpooled.wrappedBuffer(new byte[]{1, 2, 3, 4})));
            return;
        }
        ctx.channel().writeAndFlush(new TextWebSocketFrame(text));
    } else if (frame instanceof BinaryWebSocketFrame) {
        ctx.channel().writeAndFlush(frame.retain());
    } else if (frame instanceof CloseWebSocketFrame) {
        ctx.close();
    } else {
        String message = "unsupported frame type: " + frame.getClass().getName();
        throw new UnsupportedOperationException(message);
    }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
    if (logger.isLoggable(Level.FINE)) {
        logger.fine(String.format(
                "Channel %s received %s", ctx.channel().hashCode(), StringUtil.simpleClassName(frame)));
    }

    if (frame instanceof CloseWebSocketFrame) {
        handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame);
    } else if (frame instanceof PingWebSocketFrame) {
        ctx.write(new PongWebSocketFrame(frame.isFinalFragment(), frame.rsv(), frame.content()), ctx.voidPromise());
    } else if (frame instanceof TextWebSocketFrame) {
        ctx.write(frame, ctx.voidPromise());
    } else if (frame instanceof BinaryWebSocketFrame) {
        ctx.write(frame, ctx.voidPromise());
    } else if (frame instanceof ContinuationWebSocketFrame) {
        ctx.write(frame, ctx.voidPromise());
    } else if (frame instanceof PongWebSocketFrame) {
        frame.release();
        // Ignore
    } else {
        throw new UnsupportedOperationException(String.format("%s frame types not supported", frame.getClass()
                .getName()));
    }
}

private void performSend(byte[] raw) throws IOException
{
    if (this.outBuf != null)
    {
        this.outBuf.write(raw);
        raw = this.outBuf.toByteArray();
        this.outBuf = null;
    }
    //char[] encoded = Base64.encode(raw);

    if (this.binary)
    {
        this.ctx.channel().write(new BinaryWebSocketFrame(Unpooled.wrappedBuffer(raw)));
    }
    else
    {
        this.ctx.channel().write(new TextWebSocketFrame(StringUtil.toUtfString(raw)));
    }

}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    LOG.trace("NettyServerHandler: Channel write: {}", msg);
    if (isWebSocketServer() && msg instanceof ByteBuf) {
        if(isFragmentWrites()) {
            ByteBuf orig = (ByteBuf) msg;
            int origIndex = orig.readerIndex();
            int split = orig.readableBytes()/2;

            ByteBuf part1 = orig.copy(origIndex, split);
            LOG.trace("NettyServerHandler: Part1: {}", part1);
            orig.readerIndex(origIndex + split);
            LOG.trace("NettyServerHandler: Part2: {}", orig);

            BinaryWebSocketFrame frame1 = new BinaryWebSocketFrame(false, 0, part1);
            ctx.writeAndFlush(frame1);
            ContinuationWebSocketFrame frame2 = new ContinuationWebSocketFrame(true, 0, orig);
            ctx.write(frame2, promise);
        } else {
            BinaryWebSocketFrame frame = new BinaryWebSocketFrame((ByteBuf) msg);
            ctx.write(frame, promise);
        }
    } else {
        ctx.write(msg, promise);
    }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
    if (logger.isLoggable(Level.FINE)) {
        logger.fine(String.format(
                "Channel %s received %s", ctx.channel().hashCode(), StringUtil.simpleClassName(frame)));
    }

    if (frame instanceof CloseWebSocketFrame) {
        handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame);
    } else if (frame instanceof PingWebSocketFrame) {
        ctx.write(new PongWebSocketFrame(frame.isFinalFragment(), frame.rsv(), frame.content()));
    } else if (frame instanceof TextWebSocketFrame) {
        ctx.write(frame);
    } else if (frame instanceof BinaryWebSocketFrame) {
        ctx.write(frame);
    } else if (frame instanceof ContinuationWebSocketFrame) {
        ctx.write(frame);
    } else if (frame instanceof PongWebSocketFrame) {
        frame.release();
        // Ignore
    } else {
        throw new UnsupportedOperationException(String.format("%s frame types not supported", frame.getClass()
                .getName()));
    }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
  // Check for closing frame
  if (frame instanceof CloseWebSocketFrame) {
    handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
    return;
  }
  if (frame instanceof PingWebSocketFrame) {
    ctx.write(new PongWebSocketFrame(frame.content().retain()));
    return;
  }
  if (frame instanceof TextWebSocketFrame) {
    // Echo the frame
    ctx.write(frame.retain());
    return;
  }
  if (frame instanceof BinaryWebSocketFrame) {
    // Echo the frame
    ctx.write(frame.retain());
  }
}

public void handle(final Object msg) {
  ready();
  if (msg instanceof TextWebSocketFrame) {
    onTextCallback.accept(((TextWebSocketFrame) msg).text());
  } else if (msg instanceof BinaryWebSocketFrame) {
    onBinaryCallback.accept(((BinaryWebSocketFrame) msg).content().nioBuffer());
  } else if (msg instanceof CloseWebSocketFrame) {
    CloseWebSocketFrame closeFrame = ((CloseWebSocketFrame) msg).retain();
    int statusCode = closeFrame.statusCode();
    onCloseCallback.accept(statusCode == -1 ? WebSocket.NORMAL.code() : statusCode,
        Optional.ofNullable(closeFrame.reasonText()));
    handshaker.close(ctx.channel(), closeFrame).addListener(CLOSE);
  } else if (msg instanceof Throwable) {
    onErrorCallback.accept((Throwable) msg);
  }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
    if (frame instanceof CloseWebSocketFrame) {
        handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
        return;
    }
    if (frame instanceof PingWebSocketFrame) {
        ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
        return;
    }

    if (frame instanceof BinaryWebSocketFrame)
        try {
            this.connection.onMessage(((BinaryWebSocketFrame) frame).content().retain());
        } catch (Exception e) {
            logger.error("onMessage error", e);
            handshaker.close(ctx.channel(),
                    new CloseWebSocketFrame(true, 0,
                            frame.content().clear()
                                    .writeShort(1000)
                                    .writeBytes(e.getMessage().getBytes(CharsetUtil.UTF_8))
                                    .retain()));
        }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame msg) throws Exception {
  if (log.isDebugEnabled())
    log.debug("Received {} WebSocketFrame: {} from channel: {}", getTransportType().getName(), msg, ctx.channel());

  if (msg instanceof CloseWebSocketFrame) {
    sessionIdByChannel.remove(ctx.channel());
    ChannelFuture f = ctx.writeAndFlush(msg);
    f.addListener(ChannelFutureListener.CLOSE);
  } else if (msg instanceof PingWebSocketFrame) {
    ctx.writeAndFlush(new PongWebSocketFrame(msg.content()));
  } else if (msg instanceof TextWebSocketFrame || msg instanceof BinaryWebSocketFrame){
    Packet packet = PacketDecoder.decodePacket(msg.content());
    packet.setTransportType(getTransportType());
    String sessionId = sessionIdByChannel.get(ctx.channel());
    packet.setSessionId(sessionId);
    msg.release();
    ctx.fireChannelRead(packet);
  } else {
    msg.release();
    log.warn("{} frame type is not supported", msg.getClass().getName());
  }
}

@Override
protected void decode(final ChannelHandlerContext ctx, final WebSocketFrame msg, final List<Object> out)
    throws Exception {
    if (msg instanceof BinaryWebSocketFrame) {
        ByteBuf content = msg.content();
        // the content is passed to other handlers so they need to be retained.
        content.retain();
        fragments.add(content);
        if (msg.isFinalFragment()) {
            if (fragments.size() == 1) {
                out.add(fragments.get(0));
            } else {
                ByteBuf[] array = fragments.toArray(BYTE_BUF_TYPE);
                out.add(Unpooled.wrappedBuffer(array));
            }
            fragments.clear();
        }
    } else if (msg instanceof TextWebSocketFrame) {
        LOG.warn("Recieved a Websocket text frame. This was not expected. Ignoring it.");
    }
}

/**
 * 向当前客户端发送数据
 * @param message
 */
public void send(Message message) {
    byte[] bytes = message.toByteArray();

    ByteBuf b = Unpooled.buffer(bytes.length);
    b.writeBytes(bytes);

    WebSocketFrame frame = new BinaryWebSocketFrame(b);

    channel.writeAndFlush(frame);
}

public ChannelFuture write(Communication.HeshReqMessage message) {
    byte[] bytes = message.toByteArray();

    ByteBuf b = Unpooled.buffer(bytes.length);
    b.writeBytes(bytes);

    WebSocketFrame frame = new BinaryWebSocketFrame(b);
    return channel.writeAndFlush(frame);
}

@Override
protected void decode(ChannelHandlerContext ctx,
        BinaryWebSocketFrame wsFrame, List<Object> out) throws Exception {
    ByteBuf buf = wsFrame.content();

    this.messageNewDecoder.decode(ctx, buf, out);
}

@Override
protected void encode(ChannelHandlerContext ctx, Message msg,
        List<Object> out) throws Exception {
    if (msg == null || !(msg instanceof Message))
        return;

    byte[] data = ((Message) msg).toBytes();

    out.add(new BinaryWebSocketFrame(Unpooled.wrappedBuffer(data)));
}