io.netty.buffer.Unpooled#EMPTY_BUFFER源码实例Demo

@Override
public ByteBuf encode(ByteBufAllocator alloc, SpdyHeadersFrame frame) throws Exception {
    if (frame == null) {
        throw new IllegalArgumentException("frame");
    }

    if (finished) {
        return Unpooled.EMPTY_BUFFER;
    }

    ByteBuf decompressed = super.encode(alloc, frame);
    try {
        if (!decompressed.isReadable()) {
            return Unpooled.EMPTY_BUFFER;
        }

        setInput(decompressed);
        return encode(alloc);
    } finally {
        decompressed.release();
    }
}
 

@Test
public void testEmptyFullContentWithTrailer() throws Exception {
    EmbeddedChannel ch = new EmbeddedChannel(new TestEncoder());
    ch.writeInbound(new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/"));

    FullHttpResponse res = new DefaultFullHttpResponse(
            HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.EMPTY_BUFFER);
    res.trailingHeaders().set("X-Test", "Netty");
    ch.writeOutbound(res);

    Object o = ch.readOutbound();
    assertThat(o, is(instanceOf(FullHttpResponse.class)));

    res = (FullHttpResponse) o;
    assertThat(res.headers().get(Names.TRANSFER_ENCODING), is(nullValue()));

    // Content encoding shouldn't be modified.
    assertThat(res.headers().get(Names.CONTENT_ENCODING), is(nullValue()));
    assertThat(res.content().readableBytes(), is(0));
    assertThat(res.content().toString(CharsetUtil.US_ASCII), is(""));
    assertEquals("Netty", res.trailingHeaders().get("X-Test"));
    assertThat(ch.readOutbound(), is(nullValue()));
}
 

@Test
public void shouldDecodeReplicaGetResponse() {
    ByteBuf content = Unpooled.copiedBuffer("content", CharsetUtil.UTF_8);
    FullBinaryMemcacheResponse response = new DefaultFullBinaryMemcacheResponse(KEY, Unpooled.EMPTY_BUFFER,
        content.copy());
    response.setCAS(123456789L);
    response.setExtras(Unpooled.buffer().writeInt(123));
    response.setExtrasLength((byte) 4);

    ReplicaGetRequest requestMock = mock(ReplicaGetRequest.class);
    when(requestMock.bucket()).thenReturn(BUCKET);
    requestQueue.add(requestMock);
    channel.writeInbound(response);

    assertEquals(1, eventSink.responseEvents().size());
    GetResponse event = (GetResponse) eventSink.responseEvents().get(0).getMessage();
    assertEquals(123456789L, event.cas());
    assertEquals(123, event.flags());
    assertEquals("content", event.content().toString(CHARSET));
    assertEquals(BUCKET, event.bucket());
}
 

@Override
protected void channelRead0(final ChannelHandlerContext ctx, final FullHttpRequest request) {
    if (HttpMethod.CONNECT.name().equalsIgnoreCase(request.method().name())) {
        final FullHttpResponse response =
                new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.EMPTY_BUFFER);
        HttpUtil.setKeepAlive(response, true);
        HttpUtil.setContentLength(response, 0);
        if (api.getSslContext() != null) {
            final SSLEngine sslEngine = api.getSslContext().createSSLEngine();
            sslEngine.setUseClientMode(false);
            ctx.channel().pipeline().addFirst("ssl", new SslHandler(sslEngine, true));

            final String uri = request.uri();
            final String[] parts = uri.split(":");
            ctx
                    .channel()
                    .attr(BASE)
                    .set("https://" + parts[0]
                            + (parts.length > 1 && !"443".equals(parts[1]) ? ":" + parts[1] : ""));
        }
        ctx.writeAndFlush(response);
        return;
    }
    final FullHttpRequest req = request.copy(); // copy to use in a separated thread
    api.getExecutor().execute(() -> doHttpRequest(req, ctx));
}
 

@Test
public void shouldNotReleasePrependRequestContentOnRetry() throws Exception {
    ByteBuf content = Unpooled.copiedBuffer("content", CharsetUtil.UTF_8);
    FullBinaryMemcacheResponse response = new DefaultFullBinaryMemcacheResponse(KEY, Unpooled.EMPTY_BUFFER,
        content);
    response.setStatus(KeyValueStatus.ERR_NOT_MY_VBUCKET.code());

    PrependRequest requestMock = mock(PrependRequest.class);
    ByteBuf requestContent = Unpooled.copiedBuffer("content", CharsetUtil.UTF_8);
    when(requestMock.bucket()).thenReturn("bucket");
    when(requestMock.observable()).thenReturn(AsyncSubject.<CouchbaseResponse>create());
    when(requestMock.content()).thenReturn(requestContent);
    requestQueue.add(requestMock);

    assertEquals(1, content.refCnt());
    assertEquals(1, requestContent.refCnt());
    channel.writeInbound(response);
    assertEquals(1, content.refCnt());
    assertEquals(1, requestContent.refCnt());
}
 

@Test(expected = CouchbaseException.class)
public void shouldFailWhenOpaqueDoesNotMatch() throws Exception {
    ByteBuf content = Unpooled.copiedBuffer("content", CharsetUtil.UTF_8);
    FullBinaryMemcacheResponse response = new DefaultFullBinaryMemcacheResponse(KEY, Unpooled.EMPTY_BUFFER,
            content);
    response.setStatus(BinaryMemcacheResponseStatus.SUCCESS);
    response.setOpaque(1);

    PrependRequest requestMock = mock(PrependRequest.class);
    ByteBuf requestContent = Unpooled.copiedBuffer("content", CharsetUtil.UTF_8);
    when(requestMock.bucket()).thenReturn("bucket");
    AsyncSubject<CouchbaseResponse> responseSubject = AsyncSubject.<CouchbaseResponse>create();
    when(requestMock.observable()).thenReturn(responseSubject);
    when(requestMock.content()).thenReturn(requestContent);
    when(requestMock.opaque()).thenReturn(3);
    requestQueue.add(requestMock);

    channel.writeInbound(response);
    assertEquals(0, content.refCnt());
    responseSubject.toBlocking().single();
}
 

/**
 * Creates the HELLO request to ask for certain supported features.
 *
 * @param connId the connection id
 * @return the request to send over the wire
 */
private FullBinaryMemcacheRequest helloRequest(int connId) throws Exception {
    byte[] key = generateAgentJson(
        ctx.environment().userAgent(),
        ctx.coreId(),
        connId
    );
    short keyLength = (short) key.length;

    ByteBuf wanted = Unpooled.buffer(features.size() * 2);
    for (ServerFeatures feature : features) {
        wanted.writeShort(feature.value());
    }

    LOGGER.debug("Requesting supported features: {}", features);
    FullBinaryMemcacheRequest request = new DefaultFullBinaryMemcacheRequest(key, Unpooled.EMPTY_BUFFER, wanted);
    request.setOpcode(HELLO_CMD);
    request.setKeyLength(keyLength);
    request.setTotalBodyLength(keyLength + wanted.readableBytes());
    return request;
}
 

/**
 * Create a multi-mutation command.
 *
 * @param mutation the mutation type.
 * @param path     the path to mutate inside the document.
 * @param fragment the target value for the mutation. This will be released when the request is sent.
 */
@Deprecated
public MutationCommand(Mutation mutation, String path, ByteBuf fragment) {
    this.mutation = mutation;
    this.path = path;
    this.fragment = (fragment == null) ? Unpooled.EMPTY_BUFFER : fragment;
    this.expandMacros = false;
}
 

@Test
public void testParseLabel() throws RSVPParsingException {
    final LabelType output = this.simpleLabelRegistry.parseLabel(this.ctype, this.input);
    assertNotNull(output);
    assertTrue(output instanceof MockLabel);

    final ByteBuf aggregator = Unpooled.EMPTY_BUFFER;
    this.simpleLabelRegistry.serializeLabel(false, false, output, aggregator);
    Mockito.verify(this.labelSerializer).serializeLabel(false, false, output, aggregator);
}
 

@Test
public void testParseTlvs() throws BmpDeserializationException {
    final ByteBuf buffer = Unpooled.EMPTY_BUFFER;
    final TlvsBuilder builder = new TlvsBuilder();
    this.parser.parseTlvs(builder, buffer);
    assertNull(builder.getDescriptionTlv());

    this.parser.parseTlvs(builder, Unpooled.wrappedBuffer(DATA));
    assertNotNull(builder.getDescriptionTlv());
    assertEquals("test", builder.getDescriptionTlv().getDescription());
}
 

/**
 * Helper method to decode all simple subdocument response messages.
 *
 * @param request the current request.
 * @param msg the current response message.
 * @param ctx the handler context.
 * @param status the response status code.
 * @return the decoded response or null if none did match.
 */
private static CouchbaseResponse handleSubdocumentResponseMessages(BinaryRequest request, FullBinaryMemcacheResponse msg,
     ChannelHandlerContext ctx, ResponseStatus status, boolean seqOnMutation) {
    if (!(request instanceof BinarySubdocRequest))
        return null;
    BinarySubdocRequest subdocRequest = (BinarySubdocRequest) request;
    long cas = msg.getCAS();
    short statusCode = msg.getStatus();
    String bucket = request.bucket();

    MutationToken mutationToken = null;
    if (msg.getExtrasLength() > 0) {
        mutationToken = extractToken(bucket, seqOnMutation, status.isSuccess(), msg.getExtras(), request.partition());
    }

    ByteBuf fragment;
    if (msg.content() != null && msg.content().readableBytes() > 0) {
        fragment = msg.content();
    } else if (msg.content() != null) {
        msg.content().release();
        fragment = Unpooled.EMPTY_BUFFER;
    } else {
        fragment = Unpooled.EMPTY_BUFFER;
    }

    return new SimpleSubdocResponse(status, statusCode, bucket, fragment, subdocRequest, cas, mutationToken);
}
 

@Override
protected void channelRead0(ChannelHandlerContext ctx, MetricRequest m) throws Exception {
    try {
        this.dataStore.store(m.getMetric());
    } catch (TimelyException e) {
        LOG.error(e.getMessage(), e);
        this.sendHttpError(ctx, e);
        return;
    }
    FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,
            Unpooled.EMPTY_BUFFER);
    response.headers().set(HttpHeaderNames.CONTENT_TYPE, Constants.JSON_TYPE);
    response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
    sendResponse(ctx, response);
}
 

RpcData genErrorResponse(ChannelHandlerContext ctx, RpcMeta meta, boolean isExchange, int retCode) {
    RpcMeta resMeta = RpcMeta.newBuilder().setDirection(RpcMeta.Direction.RESPONSE)
            .setServiceId(meta.getServiceId()).setMsgId(meta.getMsgId()).setSequence(meta.getSequence())
            .setRetCode(retCode).build();
    if( isExchange ) {
        return new RpcData(resMeta, Unpooled.EMPTY_BUFFER);
    } else {
        Message res = serviceMetas.generateRes(meta.getServiceId(), meta.getMsgId(), retCode);
        return new RpcData(resMeta, res);
    }
}
 

@Override
public ByteBuf serializeRouteTargetConstrain(final RouteTargetConstrainChoice routeTarget) {
    final RouteTargetConstrainSerializer<RouteTargetConstrainChoice> serializer =
            this.handlers.getSerializer(routeTarget.implementedInterface());
    return serializer == null || serializer.getType() == null ? Unpooled.EMPTY_BUFFER
            : Unpooled.buffer()
            .writeByte(serializer.getType())
            .writeByte(RT_SUBTYPE)
            .writeBytes(serializer.serializeRouteTargetConstrain(routeTarget));
}
 

@Test
public void shouldDispatchFirstNMVImmediately() throws Exception {
    final CountDownLatch latch = new CountDownLatch(1);
    ClusterFacade clusterMock = mock(ClusterFacade.class);
    when(clusterMock.send(any(CouchbaseRequest.class))).then(new Answer<Object>() {
        @Override
        public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
            latch.countDown();
            return null;
        }
    });

    ConfigurationProvider providerMock = mock(ConfigurationProvider.class);
    ClusterConfig clusterConfig = mock(ClusterConfig.class);
    BucketConfig bucketConfig = mock(BucketConfig.class);
    when(providerMock.config()).thenReturn(clusterConfig);
    when(clusterConfig.bucketConfig("bucket")).thenReturn(bucketConfig);
    when(bucketConfig.hasFastForwardMap()).thenReturn(true);

    ResponseHandler handler = new ResponseHandler(ENVIRONMENT, clusterMock, providerMock);

    GetRequest request = new GetRequest("key", "bucket");
    GetResponse response = new GetResponse(ResponseStatus.RETRY, (short) 0, 0L ,0, "bucket", Unpooled.EMPTY_BUFFER,
            request);

    ResponseEvent retryEvent = new ResponseEvent();
    retryEvent.setMessage(response);
    retryEvent.setObservable(request.observable());

    handler.onEvent(retryEvent, 1, true);

    long start = System.nanoTime();
    latch.await(5, TimeUnit.SECONDS);
    long end = System.nanoTime();

    // assert immediate dispatch
    assertTrue(TimeUnit.NANOSECONDS.toMillis(end - start) < 10);
}
 

public static FullHttpRequest createGetRequest(String host, int port) {
    FullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.valueOf("HTTP/2.0"), HttpMethod.GET, "/", Unpooled.EMPTY_BUFFER);
    request.headers()
        .add(HttpHeaderNames.HOST, new String(host + ":" + port));
    request.headers()
        .add(HttpConversionUtil.ExtensionHeaderNames.SCHEME.text(), HttpScheme.HTTPS);
    request.headers()
        .add(HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP);
    request.headers()
        .add(HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.DEFLATE);
    return request;
}
 

@Override
public void decode(ByteBuf buf, ProtocolUtils.Direction direction, ProtocolVersion version) {
  this.id = ProtocolUtils.readVarInt(buf);
  this.success = buf.readBoolean();
  if (buf.isReadable()) {
    this.data = buf.readSlice(buf.readableBytes());
  } else {
    this.data = Unpooled.EMPTY_BUFFER;
  }
}
 

/**
 * Constructs a new {@link ByteBufHttpData}. Ownership of {@code buf} is taken by this
 * {@link ByteBufHttpData}, which must not be mutated anymore.
 */
public ByteBufHttpData(ByteBuf buf, boolean endOfStream) {
    length = requireNonNull(buf, "buf").readableBytes();
    if (length != 0) {
        this.buf = buf;
    } else {
        buf.release();
        this.buf = Unpooled.EMPTY_BUFFER;
    }
    this.endOfStream = endOfStream;
}
 

public DefaultHttp2UnknownFrame(byte frameType, Http2Flags flags) {
    this(frameType, flags, Unpooled.EMPTY_BUFFER);
}
 

/**
 * Creates a unsuccessful {@link MultiMutationResponse} that failed at document level.
 *
 * First error index is set to -1 and first error status is set to {@link ResponseStatus#FAILURE}.
 *
 * @param status the failed status of the request.
 * @param serverStatusCode the status code of the whole request.
 * @param bucket the bucket on which the request happened.
 * @param request the original {@link BinarySubdocMultiMutationRequest}.
 * @param cas the CAS value of the document after mutations.
 * @param mutationToken the {@link MutationToken} of the document after mutations, if available. Null otherwise.
 */
public MultiMutationResponse(ResponseStatus status, short serverStatusCode, String bucket,
                             BinarySubdocMultiMutationRequest request, long cas, MutationToken mutationToken) { //do cas and muto make sense here?
    super(status, serverStatusCode, bucket, Unpooled.EMPTY_BUFFER, request);
    this.cas = cas;
    this.mutationToken = mutationToken;
    this.firstErrorIndex = -1;
    this.firstErrorStatus = ResponseStatus.FAILURE;
    this.responses = Collections.emptyList();
}