io.netty.util.concurrent.PromiseCombiner#add ( )源码实例Demo

@Override
public void sendAsync(List<Append> appends, CompletedCallback callback) {
    Channel ch;
    try {
        checkClientConnectionClosed();
        ch = nettyHandler.getChannel();
    } catch (ConnectionFailedException e) {
        callback.complete(new ConnectionFailedException("Connection to " + connectionName + " is not established."));
        return;
    }
    PromiseCombiner combiner = new PromiseCombiner(ImmediateEventExecutor.INSTANCE);
    for (Append append : appends) {
        combiner.add(ch.write(append));
    }
    ch.flush();
    ChannelPromise promise = ch.newPromise();
    promise.addListener(future -> {
        nettyHandler.setRecentMessage();
        Throwable cause = future.cause();
        callback.complete(cause == null ? null : new ConnectionFailedException(cause));
    });
    combiner.finish(promise);
}
 

private void writeContent(ChannelHandlerContext ctx, SegmentedData data, ChannelPromise promise) {
  Headers trailingHeaders = data.trailingHeaders();
  boolean hasTrailing = trailingHeaders != null && trailingHeaders.size() > 0;
  boolean dataEos = data.endOfMessage() && !hasTrailing;

  int streamId = data.streamId();

  Http2Request request =
      Http2Request.build(streamId, new DefaultHttp2DataFrame(data.content(), dataEos), dataEos);

  if (hasTrailing) {
    Http2Headers headers = trailingHeaders.http2Headers();
    Http2Request last = Http2Request.build(streamId, headers, true);
    PromiseCombiner combiner = new PromiseCombiner();
    combiner.add(ctx.write(request, ctx.newPromise()));
    combiner.add(ctx.write(last, ctx.newPromise()));
    combiner.finish(promise);
  } else {
    ctx.write(request, promise);
  }
}
 

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
    throws Exception {
  if (msg instanceof Message) {
    Message message = (Message) msg;
    Request request = message.getRequest();
    if (request.expectsResponse()) {
      setRequest(ctx.channel().attr(KEY).get(), request);
    }

    PromiseCombiner combiner = new PromiseCombiner();
    combiner.add(write(ctx, message.getPayload()));
    combiner.add(write(ctx, message));
    combiner.finish(promise);
  } else {
    throw new RuntimeException("Only Message objects can be written to ClientCodec");
  }
}
 

private void writeRequest(ChannelHandlerContext ctx, Request request, ChannelPromise promise) {
  /*
    // TOOD(CK): define ACCEPT?
  if (!response.headers().contains(HttpHeaderNames.CONTENT_TYPE)) {
    response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");
  }
  */

  Http2Headers headers = request.headers().http2Headers();

  headers.authority(request.host()).method(request.method().asciiName()).path(request.path());

  int streamId = request.streamId();

  if (request instanceof FullRequest) {
    if (request.body().readableBytes() > 0) {
      PromiseCombiner combiner = new PromiseCombiner();
      combiner.add(ctx.write(Http2Request.build(streamId, headers, false), ctx.newPromise()));
      Http2DataFrame data = new DefaultHttp2DataFrame(request.body(), true);
      combiner.add(ctx.write(Http2Request.build(streamId, data, true), ctx.newPromise()));
      combiner.finish(promise);
    } else {
      ctx.write(Http2Request.build(streamId, headers, true), promise);
    }
  } else {
    ctx.write(Http2Request.build(streamId, headers, false), promise);
  }
}
 

@Override
public ChannelFuture writeData(final ChannelHandlerContext ctx, final int streamId, ByteBuf data, int padding,
        final boolean endOfStream, ChannelPromise promise) {
    final Http2Stream stream = connection().stream(streamId);
    final EmbeddedChannel channel = stream == null ? null : (EmbeddedChannel) stream.getProperty(propertyKey);
    if (channel == null) {
        // The compressor may be null if no compatible encoding type was found in this stream's headers
        return super.writeData(ctx, streamId, data, padding, endOfStream, promise);
    }

    try {
        // The channel will release the buffer after being written
        channel.writeOutbound(data);
        ByteBuf buf = nextReadableBuf(channel);
        if (buf == null) {
            if (endOfStream) {
                if (channel.finish()) {
                    buf = nextReadableBuf(channel);
                }
                return super.writeData(ctx, streamId, buf == null ? Unpooled.EMPTY_BUFFER : buf, padding,
                        true, promise);
            }
            // END_STREAM is not set and the assumption is data is still forthcoming.
            promise.setSuccess();
            return promise;
        }

        PromiseCombiner combiner = new PromiseCombiner();
        for (;;) {
            ByteBuf nextBuf = nextReadableBuf(channel);
            boolean compressedEndOfStream = nextBuf == null && endOfStream;
            if (compressedEndOfStream && channel.finish()) {
                nextBuf = nextReadableBuf(channel);
                compressedEndOfStream = nextBuf == null;
            }

            ChannelPromise bufPromise = ctx.newPromise();
            combiner.add(bufPromise);
            super.writeData(ctx, streamId, buf, padding, compressedEndOfStream, bufPromise);
            if (nextBuf == null) {
                break;
            }

            padding = 0; // Padding is only communicated once on the first iteration
            buf = nextBuf;
        }
        combiner.finish(promise);
    } catch (Throwable cause) {
        promise.tryFailure(cause);
    } finally {
        if (endOfStream) {
            cleanup(stream, channel);
        }
    }
    return promise;
}
 

/**
 * Note this blocks until all the channels have finished closing.
 */
public void gracefullyShutdownClientChannels()
{
    LOG.warn("Gracefully shutting down all client channels");
    try {


        // Mark all active connections to be closed after next response sent.
        LOG.warn("Flagging CLOSE_AFTER_RESPONSE on " + channels.size() + " client channels.");
        // Pick some arbitrary executor.
        PromiseCombiner closeAfterPromises = new PromiseCombiner(ImmediateEventExecutor.INSTANCE);
        for (Channel channel : channels)
        {
            ConnectionCloseType.setForChannel(channel, ConnectionCloseType.DELAYED_GRACEFUL);

            ChannelPromise closePromise = channel.pipeline().newPromise();
            channel.attr(ConnectionCloseChannelAttributes.CLOSE_AFTER_RESPONSE).set(closePromise);
            // TODO(carl-mastrangelo): remove closePromise, since I don't think it's needed.  Need to verify.
            closeAfterPromises.add(channel.closeFuture());
        }

        // Wait for all of the attempts to close connections gracefully, or max of 30 secs each.
        Promise<Void> combinedCloseAfterPromise = executor.newPromise();
        closeAfterPromises.finish(combinedCloseAfterPromise);
        combinedCloseAfterPromise.await(30, TimeUnit.SECONDS);

        // Close all of the remaining active connections.
        LOG.warn("Closing remaining active client channels.");
        List<ChannelFuture> forceCloseFutures = new ArrayList<>();
        channels.forEach(channel -> {
            if (channel.isActive()) {
                ChannelFuture f = channel.pipeline().close();
                forceCloseFutures.add(f);
            }
        });

        LOG.warn("Waiting for " + forceCloseFutures.size() + " client channels to be closed.");
        PromiseCombiner closePromisesCombiner = new PromiseCombiner(ImmediateEventExecutor.INSTANCE);
        closePromisesCombiner.addAll(forceCloseFutures.toArray(new ChannelFuture[0]));
        Promise<Void> combinedClosePromise = executor.newPromise();
        closePromisesCombiner.finish(combinedClosePromise);
        combinedClosePromise.await(5, TimeUnit.SECONDS);
        LOG.warn(forceCloseFutures.size() + " client channels closed.");
    }
    catch (InterruptedException ie) {
        LOG.warn("Interrupted while shutting down client channels");
    }
}