java类io.netty.channel.ChannelPromise的实例源码

@Override
public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise)
    throws Exception {
  if (localAddress == slowAddr) {
    // delay binding 1 second.
    eventLoop.schedule(
        () -> {
          try {
            super.bind(ctx, localAddress, promise);
          } catch (Exception e) {
            // shouldn't happen.
          }
        },
        1,
        TimeUnit.SECONDS);
  } else {
    super.bind(ctx, localAddress, promise);
  }
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    LOGGER.info("[Client ({})] <= [Server ({})] : {}",
                connectionInfo.getClientAddr(), connectionInfo.getServerAddr(),
                msg);

    if (msg instanceof HttpResponse) {
        HttpResponse response = (HttpResponse) msg;
        if (!response.headers().contains(HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text())) {
            if (streams.isEmpty()) {
                throw new IllegalStateException("No active streams");
            }
            response.headers().add(HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text(),
                                   streams.poll());
        }
    }

    ctx.write(msg, promise);
}

@Override
protected AsyncRpcClient createRpcClientRTEDuringConnectionSetup(Configuration conf) {
  setConf(conf);
  return new AsyncRpcClient(conf, new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
          ch.pipeline().addFirst(new ChannelOutboundHandlerAdapter() {
            @Override
            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
                throws Exception {
              promise.setFailure(new RuntimeException("Injected fault"));
            }
          });
        }
      });
}

@Override
public void write ( final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise ) throws Exception
{
    if ( msg instanceof OpenResponse )
    {
        writeOpenResponse ( ctx, (OpenResponse)msg, promise );
    }
    else if ( msg instanceof SyslogResponse )
    {
        writeGenericResponse ( ctx, (AbstractGenericResponse)msg, promise );
    }
    else if ( msg instanceof ServerCloseMessage )
    {
        ctx.write ( new Frame ( 0, "serverclose", (ByteBuf)null ) );
    }
    else
    {
        super.write ( ctx, msg, promise );
    }
}

protected void writeOpenResponse ( final ChannelHandlerContext ctx, final OpenResponse msg, final ChannelPromise promise )
{
    final StringBuilder sb = new StringBuilder ();

    int i = 0;
    for ( final Map.Entry<String, String> entry : msg.getOffers ().entrySet () )
    {
        if ( i > 0 )
        {
            sb.append ( Constants.LF_STRING );
        }

        sb.append ( entry.getKey () );

        if ( entry.getValue () != null )
        {
            sb.append ( '=' );
            sb.append ( entry.getValue () );
        }
        i++;
    }

    ctx.write ( createCommonResponse ( msg.getTransactionId (), msg.getCode (), msg.getMessage (), sb ) );
}

protected void processFrame ( final ChannelHandlerContext ctx, final Frame frame, final ChannelPromise promise )
{
    final int length = frame.getData () == null ? 0 : frame.getData ().readableBytes ();

    final ByteBuf data = ctx.alloc ().buffer ();
    data.writeBytes ( String.format ( "%s", frame.getTransactionId () ).getBytes ( TXNR_CHARSET ) );
    data.writeByte ( Constants.SP );
    data.writeBytes ( frame.getCommand ().getBytes ( COMMAND_CHARSET ) );
    data.writeByte ( Constants.SP );
    data.writeBytes ( String.format ( "%s", length ).getBytes ( LENGTH_CHARSET ) );

    if ( length > 0 )
    {
        data.writeByte ( Constants.SP );
        data.writeBytes ( frame.getData () );
    }

    if ( frame.getData () != null )
    {
        frame.getData ().release ();
    }

    data.writeByte ( Constants.LF );

    ctx.write ( data );
}

private void handleMessageWrite ( final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise )
{
    final ByteBuf data = encode ( ctx, msg );

    if ( data == null )
    {
        // ignore
        return;
    }

    // if the buffer is full
    if ( this.ackBuffer.isFull () )
    {
        logger.trace ( "Store message for later transmission" );
        // ... store now and re-try later
        this.messageBuffer.add ( new WriteEvent ( ctx, data, promise ) );
    }
    else
    {
        writeMessageToChannel ( ctx, data, promise );
    }
}

private void writeMessageToChannel ( final ChannelHandlerContext ctx, final ByteBuf data, final ChannelPromise promise )
{
    final int seq = this.ackBuffer.addMessage ( data );

    if ( promise == null )
    {
        ctx.write ( new InformationTransfer ( seq, this.receiveCounter, data ) );
    }
    else
    {
        ctx.write ( new InformationTransfer ( seq, this.receiveCounter, data ), promise );
    }

    logger.trace ( "Enqueued message as {} : {}", seq, data );

    // we can stop timer #2 here ... will be restarted by receive
    this.timer2.stop ();
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    if (msg instanceof DeferredMessage) {
        DeferredMessage defMsg = (DeferredMessage)msg;

        if (defMsg.isPreEncoded()) {
            // Write pre-encoded message.
            ctx.write(defMsg.encoded(), promise);
        } else {
            super.write(ctx, defMsg.encoded(), promise);
        }
    } else if (msg instanceof ByteBuf) {
        ctx.write(msg, promise);
    } else {
        super.write(ctx, msg, promise);
    }
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    IMessage message = (IMessage) msg;
    if (message.getMsgId() >= 0x40 && message.getMsgId() <= 0x4F) {
        ctx.writeAndFlush(Unpooled.wrappedBuffer(tcpMsgBuilder.buildResponseMsg((BaseTcpResponseMsg) message)));
    } else if (message.getMsgId() >= 0x60 && message.getMsgId() <= 0x6F) {
        ctx.writeAndFlush(Unpooled.wrappedBuffer(tcpMsgBuilder.buildResponseMsg((BaseTcpResponseMsg) message)));
    } else if (message.getMsgId() >= -128 && message.getMsgId() <= -113) {
        ctx.writeAndFlush(Unpooled.wrappedBuffer(tcpMsgBuilder.buildResponseMsg((BaseTcpResponseMsg) message)));
    } else if (message.getMsgId() == MsgType.INITIALIZE_DEVICE_RESPONSE_CARD) {
        ctx.writeAndFlush(Unpooled.wrappedBuffer(tcpMsgBuilder.buildInitConfirmCardNumber((TcpMsgInitResponseCardNumber) message)));
    }
    if (message.getMsgId() == 0x00) {
        ctx.writeAndFlush(Unpooled.wrappedBuffer(tcpMsgBuilder.buildResponseMsg((BaseTcpResponseMsg) message)));
    }
    if (sendListener != null) {
        sendListener.send();
    }
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    SendableMsgContainer message = (SendableMsgContainer) msg;
    ByteBuf byteBuf = ctx.alloc().buffer();
    FrameMajorHeader header = message.getFrameHeader();
    byteBuf.writeByte(FrameSetting.MAJOR_FRAME_HEAD_1);
    byteBuf.writeByte(FrameSetting.MAJOR_FRAME_HEAD_2);
    byteBuf.writeByte(header.getGroupId());
    byteBuf.writeByte(header.getMsgId());
    byteBuf.writeByte(header.getDeviceId());
    byteBuf.writeByte(header.getBackupMsg());
    byteBuf.writeShort(header.getDataLength());

    List<ByteBuf> dataList = message.getDataList();
    dataList.forEach(byteBuf::writeBytes);
    ctx.writeAndFlush(byteBuf);
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    if (!(msg instanceof Packet)) {
        throw new RuntimeException("Object is not a *Packet");
    }

    Packet packet = (Packet) msg;
    int packetId = packet.getPacketInfo().id();
    int packetSize = packet.getPacketSize() + ProtocolSizer.ofVarInt(packetId);
    int bufferSize = packetSize + ProtocolSizer.ofVarInt(packetSize);

    ByteBuf unpooledByteBuf = Unpooled.buffer(bufferSize);
    PacketSerializer serializer = new PacketSerializer(unpooledByteBuf);

    UnknownPandaServer.getLogger().info("Sending packet: " + packet.getClass().getSimpleName() + " & id: 0x" + Integer.toHexString(packetId) + " & size: " + packetSize);
    serializer.writeVarInt(packetSize);
    serializer.writeVarInt(packetId);
    packet.send(serializer);

    ctx.writeAndFlush(serializer, promise);
}

public void write(ChannelHandlerContext context, Object packet, ChannelPromise promise) throws Exception
{
    BaseComponent[] components = interceptor.getComponents(packet);
    if(components != null)
    {
        boolean allowed = isAllowed(components);
        boolean paused = isPaused();
        if(!paused || !allowed)
        {
            while(messageQueue.size() > 20)
                messageQueue.remove();
            messageQueue.add(components);
        }

        if(paused && !allowed)
            return;
    }

    super.write(context, packet, promise);
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception
{
    if (msg instanceof FMLProxyPacket)
    {
        if (side == Side.CLIENT)
        {
            //Client to server large packets are not supported to prevent client being bad.
            ctx.write(((FMLProxyPacket) msg).toC17Packet(), promise);
        }
        else
        {
            List<Packet<INetHandlerPlayClient>> parts = ((FMLProxyPacket)msg).toS3FPackets();
            for (Packet<INetHandlerPlayClient> pkt : parts)
            {
                ctx.write(pkt, promise);
            }
        }
    }
    else
    {
        ctx.write(msg, promise);
    }
}

@Override
public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
    final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
    log.info("NETTY CLIENT PIPELINE: DISCONNECT {}", remoteAddress);
    closeChannel(ctx.channel());
    super.disconnect(ctx, promise);

    if (NettyRemotingClient.this.channelEventListener != null) {
        NettyRemotingClient.this.putNettyEvent(new NettyEvent(NettyEventType.CLOSE, remoteAddress.toString(), ctx.channel()));
    }
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    try {
        msg = onPacketOutAsync(player, ctx.channel(), msg);
    } catch (Exception e) {
        plugin.getLogger().log(Level.SEVERE, "Error in onPacketOutAsync().", e);
    }

    if (msg != null) {
        super.write(ctx, msg, promise);
    }
}

@Override
public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress,
        SocketAddress localAddress, ChannelPromise promise) throws Exception {
    final String local = localAddress == null ? "UNKNOW" : localAddress.toString();
    final String remote = remoteAddress == null ? "UNKNOW" : remoteAddress.toString();
    log.info("NETTY CLIENT PIPELINE: CONNECT  {} => {}", local, remote);
    super.connect(ctx, remoteAddress, localAddress, promise);
}

@Override
public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
    final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
    log.info("NETTY CLIENT PIPELINE: DISCONNECT {}", remoteAddress);
    closeChannel(ctx.channel());
    super.disconnect(ctx, promise);
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    if (acceptOutboundMessage(msg)) {
        if (!promise.setUncancellable()) {
            return;
        }
    }

    try {
        super.write(ctx, msg, promise);
    } catch (Exception e) {
        promise.tryFailure(e);
        throw e;
    }
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    if (msg instanceof QueueCommand) {
        QueueCommand data = (QueueCommand) msg;
        QueueCommandHolder holder = queue.peek();
        if (holder != null && holder.getCommand() == data) {
            super.write(ctx, msg, promise);
        } else {
            queue.add(new QueueCommandHolder(data, promise));
            sendData(ctx.channel());
        }
    } else {
        super.write(ctx, msg, promise);
    }
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    if (acceptOutboundMessage(msg)) {
        if (!promise.setUncancellable()) {
            return;
        }
    }

    super.write(ctx, msg, promise);
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    if (msg instanceof FullHttpRequest) {
        LOGGER.info("[Client ({})] => [Server ({})] : (PENDING) {}",
                    connectionInfo.getClientAddr(), connectionInfo.getServerAddr(),
                    msg);
        HttpRequest request = (HttpRequest) msg;
        pendings.offer(new RequestPromise(request, promise));
        next();
    } else if (msg instanceof HttpObject) {
        throw new IllegalStateException("Cannot handled message: " + msg.getClass());
    } else {
        ctx.write(msg, promise);
    }
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    synchronized (pendings) {
        pendings.add(msg);
    }
    if (ctx.isRemoved()) {
        flushPendings(ctx);
        ctx.flush();
    }
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    LOGGER.info("[Client ({})] => [Server ({})] : {}",
                connectionInfo.getClientAddr(), connectionInfo.getServerAddr(),
                msg);
    if (msg instanceof FullHttpRequest) {
        HttpMessage httpMessage = (HttpRequest) msg;
        httpMessage.headers().add(ExtensionHeaderNames.SCHEME.text(), "https");
    } else if (msg instanceof HttpObject) {
        throw new IllegalStateException("Cannot handle message: " + msg.getClass());
    }

    ctx.writeAndFlush(msg, promise);
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    if (msg instanceof ByteBuf && transportServiceAdapter != null) {
        // record the number of bytes send on the channel
        promise.addListener(f -> transportServiceAdapter.addBytesSent(((ByteBuf) msg).readableBytes()));
    }
    ctx.write(msg, promise);
}

@Override
public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception
{
    if (side == Side.CLIENT)
    {
        MinecraftForge.EVENT_BUS.post(new FMLNetworkEvent.ClientDisconnectionFromServerEvent(manager));
    }
    else
    {
        MinecraftForge.EVENT_BUS.post(new FMLNetworkEvent.ServerDisconnectionFromClientEvent(manager));
    }
    cleanAttributes(ctx);
    ctx.disconnect(promise);
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    outPackets.incrementAndGet();
    if (msg instanceof ByteBuf) {
        outSize.addAndGet(((ByteBuf) msg).readableBytes());
    } else if (msg instanceof DatagramPacket) {
        outSize.addAndGet(((DatagramPacket) msg).content().readableBytes());
    }
    super.write(ctx, msg, promise);
}

@Override
public void write(ChannelHandlerContext ctx, Object message, ChannelPromise promise)
        throws Exception
{
    if (message instanceof ThriftRequest) {
        ThriftRequest thriftRequest = (ThriftRequest) message;
        sendMessage(ctx, thriftRequest, promise);
    }
    else {
        ctx.write(message, promise);
    }
}

private void sendMessage(ChannelHandlerContext context, ThriftRequest thriftRequest, ChannelPromise promise)
        throws Exception
{
    // todo ONEWAY_SEQUENCE_ID is a header protocol thing... make sure this works with framed and unframed
    int sequenceId = thriftRequest.isOneway() ? ONEWAY_SEQUENCE_ID : this.sequenceId.incrementAndGet();
    RequestHandler requestHandler = new RequestHandler(thriftRequest, sequenceId);

    // register timeout
    requestHandler.registerRequestTimeout(context.executor());

    // write request
    ByteBuf requestBuffer = requestHandler.encodeRequest(context.alloc());

    // register request if we are expecting a response
    if (!thriftRequest.isOneway()) {
        if (pendingRequests.putIfAbsent(sequenceId, requestHandler) != null) {
            requestHandler.onChannelError(new TTransportException("Another request with the same sequenceId is already in progress"));
        }
    }

    try {
        ChannelFuture sendFuture = context.write(requestBuffer, promise);
        sendFuture.addListener(future -> messageSent(context, sendFuture, requestHandler));
    }
    catch (Throwable t) {
        onError(context, t);
    }
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    try {
        msg = onPacketOutAsync(player, ctx.channel(), msg);
    } catch (Exception e) {
        plugin.getLogger().log(Level.SEVERE, "Error in onPacketOutAsync().", e);
    }

    if (msg != null) {
        super.write(ctx, msg, promise);
    }
}