Java 类io.netty.util.concurrent.PromiseCombiner 实例源码

@Override
public void sendAsync(List<Append> appends, CompletedCallback callback) {
    recentMessage.set(true);
    Channel ch;
    try {
        ch = getChannel();
    } catch (ConnectionFailedException e) {
        callback.complete(new ConnectionFailedException("Connection to " + connectionName + " is not established."));
        return;
    }
    PromiseCombiner combiner = new PromiseCombiner();
    for (Append append : appends) {
        batchSizeTracker.recordAppend(append.getEventNumber(), append.getData().readableBytes());
        combiner.add(ch.write(append));
    }
    ch.flush();
    ChannelPromise promise = ch.newPromise();
    promise.addListener(new GenericFutureListener<Future<? super Void>>() {
        @Override
        public void operationComplete(Future<? super Void> future) throws Exception {
            Throwable cause = future.cause();
            callback.complete(cause == null ? null : new ConnectionFailedException(cause));
        }
    });
    combiner.finish(promise);
}

void writeContent(ChannelHandlerContext ctx, StreamingData data, ChannelPromise promise) {
  int streamId = 0; // TODO(CK): need a no stream constant somewhere
  boolean dataEos = data.endOfStream() && data.trailingHeaders().size() == 0;
  Http2Request request =
      Http2Request.build(streamId, new DefaultHttp2DataFrame(data.content(), dataEos), dataEos);

  if (data.trailingHeaders().size() != 0) {
    Http2Headers headers = data.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);
  }
}

void writeContent(ChannelHandlerContext ctx, StreamingData data, ChannelPromise promise) {
  Request request = getChannelRequest(ctx);
  int streamId = request.streamId();
  if (data.endOfStream()) {
    setChannelRequest(ctx, null);
  }

  boolean dataEos = data.endOfStream() && data.trailingHeaders().size() == 0;
  Http2Response response =
      Http2Response.build(streamId, new DefaultHttp2DataFrame(data.content(), dataEos), dataEos);

  if (data.trailingHeaders().size() != 0) {
    Http2Headers headers = data.trailingHeaders().http2Headers();
    Http2Response last = Http2Response.build(streamId, headers, true);
    PromiseCombiner combiner = new PromiseCombiner();
    combiner.add(ctx.write(response, ctx.newPromise()));
    combiner.add(ctx.write(last, ctx.newPromise()));
    combiner.finish(promise);
  } else {
    ctx.write(response, 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");
  }
}

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);
  }
}

public ChannelFuture write(Request request) {
  if (channel == null) {
    ChannelFuture future = connect();
    channel = future.channel();
    ChannelPromise promise = channel.newPromise();
    PromiseCombiner combiner = new PromiseCombiner();
    combiner.add(future.addListener(connectionListener));
    combiner.add(channel.writeAndFlush(request).addListener(writeListener));
    combiner.finish(promise);
    return promise;
  } else {
    return channel.writeAndFlush(request).addListener(writeListener);
  }
}

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

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