java.nio.channels.NotYetConnectedException#io.netty.util.ReferenceCountUtil源码实例Demo

@Override
public Subscriber<? super I> call(final Subscriber<? super I> subscriber) {
    return new Subscriber<I>() {
        @Override
        public void onCompleted() {
            subscriber.onCompleted();
        }

        @Override
        public void onError(Throwable e) {
            subscriber.onError(e);
        }

        @Override
        public void onNext(I t) {
            try {
                subscriber.onNext(t);
            } finally {
                ReferenceCountUtil.release(t);
            }
        }
    };
}
 

/**
 * Propagate failures to all buffered writes.
 */
@SuppressWarnings("FutureReturnValueIgnored")
protected final void fail(ChannelHandlerContext ctx, Throwable cause) {
  if (failCause == null) {
    failCause = cause;
  }
  if (bufferedWrites != null) {
    while (!bufferedWrites.isEmpty()) {
      ChannelWrite write = bufferedWrites.poll();
      write.promise.setFailure(cause);
      ReferenceCountUtil.release(write.msg);
    }
    bufferedWrites = null;
  }

  // In case something goes wrong ensure that the channel gets closed as the
  // NettyClientTransport relies on the channel's close future to get completed.
  ctx.close();
}
 

@Test
public void testDecode() throws Exception {
    EmbeddedChannel ch = new EmbeddedChannel(
            new DelimiterBasedFrameDecoder(8192, true, Delimiters.lineDelimiter()));

    ch.writeInbound(Unpooled.copiedBuffer("first\r\nsecond\nthird", CharsetUtil.US_ASCII));

    ByteBuf buf = ch.readInbound();
    assertEquals("first", buf.toString(CharsetUtil.US_ASCII));

    ByteBuf buf2 = ch.readInbound();
    assertEquals("second", buf2.toString(CharsetUtil.US_ASCII));
    assertNull(ch.readInbound());
    ch.finish();

    ReferenceCountUtil.release(ch.readInbound());

    buf.release();
    buf2.release();
}
 

@Override
protected void channelRead0(ChannelHandlerContext ctx, Object o) throws Exception {

    Channel ch = ctx.channel();

    if (!handshaker.isHandshakeComplete()) {
        handshaker.finishHandshake(ch, (FullHttpResponse) o);
        // the handshake response was processed upgrade is complete
        handshakeLatch.countDown();
        ReferenceCountUtil.release(o);
        return;
    }

    if (o instanceof FullHttpResponse) {
        FullHttpResponse response = (FullHttpResponse) o;
        ReferenceCountUtil.release(o);
        throw new Exception("Unexpected HttpResponse (status=" + response.getStatus() + ", content="
                + response.content().toString(CharsetUtil.UTF_8) + ')');
    }
    ctx.fireChannelRead(o);
}
 

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  boolean needRelease = true;
  try {
    if (msg instanceof FullHttpRequest) {
      FullHttpRequest request = (FullHttpRequest) msg;
      messageReceived(ctx, request);
    } else {
      needRelease = false;
      ctx.fireChannelRead(msg);
    }
  } finally {
    if (needRelease) {
      ReferenceCountUtil.release(msg);
    }
  }
}
 

/**
 * Will remove the current message, mark its {@link ChannelPromise} as success and return {@code true}. If no
 * flushed message exists at the time this method is called it will return {@code false} to signal that no more
 * messages are ready to be handled.
 */
public boolean remove() {
    Entry e = flushedEntry;
    if (e == null) {
        return false;
    }
    Object msg = e.msg;

    ChannelPromise promise = e.promise;
    int size = e.pendingSize;

    removeEntry(e);

    if (!e.cancelled) {
        // only release message, notify and decrement if it was not canceled before.
        ReferenceCountUtil.safeRelease(msg);
        safeSuccess(promise);
        decrementPendingOutboundBytes(size, false);
    }

    // recycle the entry
    e.recycle();

    return true;
}
 

private ByteBuf mergeNow() {
    int size = now.size();

    if (size == 1) {
        return now.get(0);
    }

    int i = 0;
    CompositeByteBuf result = allocator.compositeBuffer(size);

    try {
        for (; i < size; ++i) {
            result.addComponent(true, now.get(i));
        }

        return result;
    } catch (Throwable e) {
        ReferenceCountUtil.safeRelease(result);

        for (; i < size; ++i) {
            ReferenceCountUtil.safeRelease(now.get(i));
        }

        throw e;
    }
}
 

@VisibleForTesting
static Set<String> supportedProtocols(SslContextBuilder builder) {
    SslContext ctx = null;
    SSLEngine engine = null;
    try {
        ctx = builder.build();
        engine = ctx.newEngine(PooledByteBufAllocator.DEFAULT);
        return ImmutableSet.copyOf(engine.getSupportedProtocols());
    } catch (Exception e) {
        throw new IllegalStateException(
                "Failed to get the list of supported protocols from an SSLContext.", e);
    } finally {
        ReferenceCountUtil.release(engine);
        ReferenceCountUtil.release(ctx);
    }
}
 

private ByteBuf mergeNow() {
    int size = now.size();

    if (size == 1) {
        return now.get(0);
    }

    int i = 0;
    CompositeByteBuf result = allocator.compositeBuffer(size);

    try {
        for (; i < size; ++i) {
            result.addComponent(true, now.get(i));
        }

        return result;
    } catch (Throwable e) {
        ReferenceCountUtil.safeRelease(result);

        for (; i < size; ++i) {
            ReferenceCountUtil.safeRelease(now.get(i));
        }

        throw e;
    }
}
 

@Test
public void testPipelinedSpdyDataFrames() throws Exception {
    int streamId1 = RANDOM.nextInt() & 0x7FFFFFFF | 0x01;
    int streamId2 = RANDOM.nextInt() & 0x7FFFFFFF | 0x01;
    byte flags = 0;
    int length = 0;

    ByteBuf buf = ReferenceCountUtil.releaseLater(Unpooled.buffer(2 * (SPDY_HEADER_SIZE + length)));
    encodeDataFrameHeader(buf, streamId1, flags, length);
    encodeDataFrameHeader(buf, streamId2, flags, length);

    delegate.readDataFrame(streamId1, false, Unpooled.EMPTY_BUFFER);
    delegate.readDataFrame(streamId2, false, Unpooled.EMPTY_BUFFER);
    replay(delegate);
    decoder.decode(buf);
    verify(delegate);
    assertFalse(buf.isReadable());
}
 

private static FieldValue retainedLargeField(List<ByteBuf> parts) {
    int i;
    int successSentinel = 0;
    int size = parts.size();

    try {
        for (i = 0; i < size; ++i) {
            parts.get(i).retain();
            successSentinel = i + 1;
        }

        return new LargeFieldValue(parts);
    } catch (Throwable e) {
        if (successSentinel < size) {
            // Retains failed, even not call `FieldValue.of`.
            // So release all retained buffers.
            // Of course, this still does not solve call-stack
            // overflow when calling `FieldValue.of`.
            for (i = 0; i < successSentinel; ++i) {
                ReferenceCountUtil.safeRelease(parts.get(i));
            }
        }

        throw e;
    }
}
 

private ChannelFuture doWriteSplitData(int id, HttpData data, boolean endStream) {
    try {
        int offset = 0;
        int remaining = data.length();
        ChannelFuture lastFuture;
        for (;;) {
            // Ensure an HttpContent does not exceed the maximum length of a cleartext TLS record.
            final int chunkSize = Math.min(MAX_TLS_DATA_LENGTH, remaining);
            lastFuture = write(id, new DefaultHttpContent(dataChunk(data, offset, chunkSize)), false);
            remaining -= chunkSize;
            if (remaining == 0) {
                break;
            }
            offset += chunkSize;
        }

        if (endStream) {
            lastFuture = write(id, LastHttpContent.EMPTY_LAST_CONTENT, true);
        }

        ch.flush();
        return lastFuture;
    } finally {
        ReferenceCountUtil.safeRelease(data);
    }
}
 

public static Blob createBlob(List<ByteBuf> value) {
    int size = value.size(), i = 0;

    try {
        for (; i < size; ++i) {
            value.get(i).retain();
        }

        return new MultiBlob(value);
    } catch (Throwable e) {
        for (int j = 0; j < i; ++j) {
            ReferenceCountUtil.safeRelease(value.get(j));
        }

        throw e;
    }
}
 

/**
 * Test that a previously inserted document is replicated to at least one replica.
 *
 * This test assumes a cluster setup where at least one replica is configured on the bucket and more or equal
 * to two nodes are available in order to correctly complete the test.
 */
@Test
public void shouldObserveReplicateToOne() {
    Assume.assumeTrue(numberOfReplicas >= 1 && numberOfNodes >= 2);

    InsertRequest request = new InsertRequest("persInsDoc2", Unpooled.copiedBuffer("test", CharsetUtil.UTF_8), bucket());
    InsertResponse response = cluster().<InsertResponse>send(request).toBlocking().single();
    assertTrue(response.status().isSuccess());
    ReferenceCountUtil.release(response);

    Boolean observeSuccess = Observe.call(
            cluster(),
            bucket(),
            "persInsDoc2",
            response.cas(),
            false,
            Observe.PersistTo.NONE,
            Observe.ReplicateTo.ONE,
            BestEffortRetryStrategy.INSTANCE
    ).timeout(5, TimeUnit.SECONDS).toBlocking().single();

    assertTrue(observeSuccess);
}
 

private static void checkSimpleAuthMetadataEncoding(
    String username, String password, int usernameLength, int passwordLength, ByteBuf byteBuf) {
  Assertions.assertThat(byteBuf.capacity())
      .isEqualTo(AUTH_TYPE_ID_LENGTH + USER_NAME_BYTES_LENGTH + usernameLength + passwordLength);

  Assertions.assertThat(byteBuf.readUnsignedByte() & ~0x80)
      .isEqualTo(WellKnownAuthType.SIMPLE.getIdentifier());
  Assertions.assertThat(byteBuf.readUnsignedByte()).isEqualTo((short) usernameLength);

  Assertions.assertThat(byteBuf.readCharSequence(usernameLength, CharsetUtil.UTF_8))
      .isEqualTo(username);
  Assertions.assertThat(byteBuf.readCharSequence(passwordLength, CharsetUtil.UTF_8))
      .isEqualTo(password);

  ReferenceCountUtil.release(byteBuf);
}
 

@Override
public @NotNull Mono<Void> metadataPush(@NotNull Payload payload) {
    try {
        if (payload.metadata().readableBytes() > 0) {
            CloudEventImpl<?> cloudEvent = Json.decodeValue(payload.getMetadataUtf8(), CLOUD_EVENT_TYPE_REFERENCE);
            //todo
            String type = cloudEvent.getAttributes().getType();
            if (UpstreamClusterChangedEvent.class.getCanonicalName().equalsIgnoreCase(type)) {
                handleUpstreamClusterChangedEvent(cloudEvent);
            }
        }
    } catch (Exception e) {
        log.error(RsocketErrorCode.message(RsocketErrorCode.message("RST-610500", e.getMessage())), e);
    } finally {
        ReferenceCountUtil.safeRelease(payload);
    }
    return Mono.empty();
}
 

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    if (dstChannel.isActive()) {
        ByteBuf byteBuf = (ByteBuf) msg;
        try {
            if (!byteBuf.hasArray()) {
                byte[] bytes = new byte[byteBuf.readableBytes()];
                byteBuf.getBytes(0, bytes);
                if (uplink) {
                    bytes = wrapper.unwrap(bytes);
                    if (bytes != null) {
                        dstChannel.writeAndFlush(Unpooled.wrappedBuffer(bytes));
                        log.info("\tClient ==========> Target \tSend [{} bytes]", bytes.length);
                    }
                } else {
                    dstChannel.writeAndFlush(Unpooled.wrappedBuffer(wrapper.wrap(bytes)));
                    log.info("\tClient <========== Target \tGet [{} bytes]", bytes.length);
                }
            }
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }
}
 

@Override
public RspSettlementInfo parseFrom(ByteBuf body, RspError error) {
	try {
		RspSettlementInfo info = new RspSettlementInfo();
		byte[] tradingDay = new byte[9];
		body.readBytes(tradingDay);
		info.setTradingDay(StringUtils.trimToEmpty(new String(tradingDay)));
		info.setSettlementID(body.readInt());
		byte[] brokerID = new byte[11];
		body.readBytes(brokerID);
		info.setBrokerID(StringUtils.trimToEmpty(new String(brokerID)));
		byte[] investorID = new byte[13];
		body.readBytes(investorID);
		info.setInvestorID(StringUtils.trimToEmpty(new String(investorID)));
		info.setSequenceNo(body.readInt());
		byte[] content = new byte[501];
		body.readBytes(content);
		info.setContent(content);
		return info;
	} finally {
		ReferenceCountUtil.release(body);
	}
}
 

/**
 * Mark the headers sent
 *
 * @return true if marked for the first time
 */
protected final boolean markSentHeaders(Object... objectsToRelease) {
	try {
		if (!hasSentHeaders()) {
			beforeMarkSentHeaders();
		}
	}
	catch (RuntimeException e) {
		for (Object o : objectsToRelease) {
			try {
				ReferenceCountUtil.release(o);
			}
			catch (Throwable e2) {
				// keep going
			}
		}
		throw e;
	}
	return HTTP_STATE.compareAndSet(this, READY, HEADERS_SENT);
}
 

@Override
public final void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (finished) {
        // Received a message after the connection has been established; pass through.
        suppressChannelReadComplete = false;
        ctx.fireChannelRead(msg);
    } else {
        suppressChannelReadComplete = true;
        Throwable cause = null;
        try {
            boolean done = handleResponse(ctx, msg);
            if (done) {
                setConnectSuccess();
            }
        } catch (Throwable t) {
            cause = t;
        } finally {
            ReferenceCountUtil.release(msg);
            if (cause != null) {
                setConnectFailure(cause);
            }
        }
    }
}
 

@Override
public void channelRead0(ChannelHandlerContext ctx, String message) throws Exception {
    try {
        long currentTimestampInMilliseconds = System.currentTimeMillis();
        
        List<GraphiteMetric> graphiteMetrics = GraphiteMetric.parseGraphiteMetrics(message, 
                GlobalVariables.graphiteAggregatedPrefix, currentTimestampInMilliseconds);
        
        for (GraphiteMetric graphiteMetric : graphiteMetrics) {
            long hashKey = GlobalVariables.metricHashKeyGenerator.incrementAndGet();
            graphiteMetric.setHashKey(hashKey);
            if (graphiteMetric.getMetricKey() != null) graphiteMetric.getMetricKey().hashCode();
            GlobalVariables.graphiteAggregatorMetrics.put(graphiteMetric.getHashKey(), graphiteMetric);
            GlobalVariables.incomingMetricsCount.incrementAndGet();
        }
        
        if (ApplicationConfiguration.isDebugModeEnabled()) {
            logger.info("TCP_Graphite_Aggregator_Received_Metrics=" + graphiteMetrics.size());
            logger.info("TCP_Graphite_Aggregator_String=\"" + message + "\"");
        }
    }
    finally {
        ReferenceCountUtil.release(message);
    }
}
 

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    if (dstChannel.isActive()) {
        ByteBuf byteBuf = (ByteBuf) msg;
        try {
            if (!byteBuf.hasArray()) {
                byte[] bytes = new byte[byteBuf.readableBytes()];
                byteBuf.getBytes(0, bytes);
                if (uplink) {
                    dstChannel.writeAndFlush(Unpooled.wrappedBuffer(wrapper.wrap(bytes)));
                    log.info("\tClient ==========> Target \tSend [{} bytes]", bytes.length);
                } else {
                    bytes = wrapper.unwrap(bytes);
                    if (bytes != null) {
                        dstChannel.writeAndFlush(Unpooled.wrappedBuffer(bytes));
                        log.info("\tClient <========== Target \tGet [{} bytes]", bytes.length);
                    }
                }
            }
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }
}
 

public void updateIncoming(float delta) {
  cache.clear();
  int cached = inPackets.drainTo(cache);
  if (DEBUG_RECEIVED_CACHE && cached > 0) Gdx.app.debug(TAG, "processing " + cached + " packets...");
  for (D2GSPacket packet : cache) {
    if (DEBUG_RECEIVED_PACKETS && !ignoredPackets.get(packet.data.dataType())) Gdx.app.debug(TAG, "processing " + packet + " packet from " + packet.from);
    try {
      packet.id = ids.get(packet.from, INVALID_CLIENT);
      if (packet.id == INVALID_CLIENT && packet.dataType != D2GSData.Connection) {
        Gdx.app.error(TAG, "  " + packet + "from invalid client and not a connection request");
        continue;
      }
      processPacket(packet);
    } finally {
      ReferenceCountUtil.release(packet.bb);
    }
  }
}
 

@Override
public ForwardCloseResponse decodeResponse(ByteBuf buffer) throws CipResponseException, PartialResponseException {
    MessageRouterResponse mResponse = MessageRouterResponse.decode(buffer);

    int generalStatus = mResponse.getGeneralStatus();

    try {
        if (generalStatus == 0x00) {
            return ForwardCloseResponse.decode(mResponse.getData());
        } else {
            throw new CipResponseException(generalStatus, mResponse.getAdditionalStatus());
        }
    } finally {
        ReferenceCountUtil.release(mResponse.getData());
    }
}
 

/**
 * Writes the specified {@link HttpObject} to {@link DecodedHttpResponse}. This method is only called
 * from {@link Http1ResponseDecoder} and {@link Http2ResponseDecoder}. If this returns {@code false},
 * it means the response stream has been closed due to disconnection or by the response consumer.
 * So the caller do not need to handle such cases because it will be notified to the response
 * consumer anyway.
 */
@Override
public boolean tryWrite(HttpObject o) {
    boolean wrote = false;
    switch (state) {
        case WAIT_NON_INFORMATIONAL:
            wrote = handleWaitNonInformational(o);
            break;
        case WAIT_DATA_OR_TRAILERS:
            wrote = handleWaitDataOrTrailers(o);
            break;
        case DONE:
            ReferenceCountUtil.safeRelease(o);
            break;
    }

    return wrote;
}
 

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  if (DEBUG_CALLS) Gdx.app.debug(TAG, "channelRead");
  boolean release = true;
  try {
    if (accept(msg)) {
      @SuppressWarnings("unchecked")
      T castedMsg = (T) msg;
      messageReceived(ctx, castedMsg);
    } else {
      release = false;
      ctx.fireChannelRead(msg);
    }
  } finally {
    if (release) ReferenceCountUtil.release(msg);
  }
}
 

@Override
@NotNull
public ByteBuf encodingResult(@Nullable Object result) throws EncodingException {
    if (result != null) {
        ByteBuf byteBuf = PooledByteBufAllocator.DEFAULT.buffer();
        try {
            ByteBufOutputStream bos = new ByteBufOutputStream(byteBuf);
            objectMapper.writeValue((OutputStream) bos, result);
            return byteBuf;
        } catch (Exception e) {
            ReferenceCountUtil.safeRelease(byteBuf);
            throw new EncodingException(RsocketErrorCode.message("RST-700500", result.toString(), "Bytebuf"), e);
        }
    }
    return EMPTY_BUFFER;
}
 

@Test
public void testDecodeWithStrip() throws Exception {
    EmbeddedChannel ch = new EmbeddedChannel(new LineBasedFrameDecoder(8192, true, false));

    ch.writeInbound(copiedBuffer("first\r\nsecond\nthird", CharsetUtil.US_ASCII));

    ByteBuf buf = ch.readInbound();
    assertEquals("first", buf.toString(CharsetUtil.US_ASCII));

    ByteBuf buf2 = ch.readInbound();
    assertEquals("second", buf2.toString(CharsetUtil.US_ASCII));
    assertNull(ch.readInbound());
    ch.finish();

    ReferenceCountUtil.release(ch.readInbound());

    buf.release();
    buf2.release();
}
 

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  if (msg instanceof HttpRequest) {
    final HttpRequest req = (HttpRequest) msg;
    final HttpMethod requestMethod = req.method();
    final QueryStringDecoder queryDecoder = new QueryStringDecoder(req.uri());
    final String requestPath = queryDecoder.path();

    boolean disconnect = queryDecoder.parameters().containsKey(DISCONNECT);
    if (disconnect) {
      if (log.isDebugEnabled())
        log.debug("Received HTTP disconnect request: {} {} from channel: {}", requestMethod, requestPath, ctx.channel());

      final String sessionId = PipelineUtils.getSessionId(requestPath);
      final Packet disconnectPacket = new Packet(PacketType.DISCONNECT, sessionId);
      disconnectPacket.setOrigin(PipelineUtils.getOrigin(req));
      ctx.fireChannelRead(disconnectPacket);
      ReferenceCountUtil.release(msg);
      return;
    }
  }
  ctx.fireChannelRead(msg);
}
 

@Override
public void beforeRequest(Channel clientChannel, HttpContent httpContent,
    HttpProxyInterceptPipeline pipeline) throws Exception {
  if (!matchFlag) {
    super.beforeRequest(clientChannel, httpContent, pipeline);
    return;
  }
  if (content != null) {
    ByteBuf temp = httpContent.content().slice();
    content.writeBytes(temp);
    if (httpContent instanceof LastHttpContent) {
      try {
        byte[] contentBts = new byte[content.readableBytes()];
        content.readBytes(contentBts);
        ((HttpRequestInfo) pipeline.getHttpRequest()).setContent(contentBts);
      } finally {
        ReferenceCountUtil.release(content);
      }
    }
  }
  pipeline.beforeRequest(clientChannel, httpContent);
}