java.security.CryptoPrimitive#io.netty.buffer.UnpooledByteBufAllocator源码实例Demo

/**
 * 反转义
 */
@Override
public ByteBuf unEscape(ByteBuf source) {
    int low = source.readerIndex();
    int high = source.writerIndex();

    int mark = source.indexOf(low, high, (byte) 0x7d);
    if (mark == -1)
        return source;

    List<ByteBuf> bufList = new ArrayList<>(3);

    int len;
    do {

        len = mark + 2 - low;
        bufList.add(slice(source, low, len));
        low += len;

        mark = source.indexOf(low, high, (byte) 0x7d);
    } while (mark > 0);

    bufList.add(source.slice(low, high - low));

    return new CompositeByteBuf(UnpooledByteBufAllocator.DEFAULT, false, bufList.size(), bufList);
}
 

private ByteBuf bufferFor(
    com.google.protobuf.GeneratedMessageV3 protoObject, boolean compressed) {
  byte[] dataBytes = protoObject.toByteArray();
  int length = dataBytes.length;
  byte[] lengthByteBuffer = ByteBuffer.allocate(4).putInt(length).array();
  int compressedFlag = compressed ? 1 : 0;
  byte[] compressedByteBuffer = ByteBuffer.allocate(1).put((byte) compressedFlag).array();

  ByteBuf grpcRequestBuffer = UnpooledByteBufAllocator.DEFAULT.buffer(length + 5, length + 5);

  grpcRequestBuffer.writeBytes(compressedByteBuffer);
  grpcRequestBuffer.writeBytes(lengthByteBuffer);
  grpcRequestBuffer.writeBytes(dataBytes);

  return grpcRequestBuffer;
}
 

@Test
public void usingNettyDataBufferFactory_HttpData() {
    final DataBufferFactoryWrapper<?> wrapper =
            new DataBufferFactoryWrapper<>(new NettyDataBufferFactory(UnpooledByteBufAllocator.DEFAULT));

    final HttpData httpData1 = HttpData.ofUtf8("abc");

    final DataBuffer buffer = wrapper.toDataBuffer(httpData1);
    assertThat(buffer).isInstanceOf(NettyDataBuffer.class);
    assertThat(((NettyDataBuffer) buffer).getNativeBuffer().refCnt()).isOne();

    final HttpData httpData2 = wrapper.toHttpData(buffer);
    assertThat(httpData2).isInstanceOf(PooledHttpData.class);
    assertThat(((PooledHttpData) httpData2).content())
            .isEqualTo(((NettyDataBuffer) buffer).getNativeBuffer());
    assertThat(((PooledHttpData) httpData2).refCnt()).isOne();
}
 

@Test
public void testByteBuf() {
    ByteBuf byteBuf = UnpooledByteBufAllocator.DEFAULT.buffer();
    System.out.println("readable bytes: " + byteBuf.readableBytes());

    byteBuf.writeBytes(new byte[]{'+', 'P', 'R', 'O', 'X', 'Y'});

    System.out.println("reader index: " + byteBuf.readerIndex());
    System.out.println("readable bytes: " + byteBuf.readableBytes());

    byteBuf.readByte();
    byteBuf.readByte();

    System.out.println("reader index: " + byteBuf.readerIndex());
    System.out.println("readable bytes: " + byteBuf.readableBytes());
}
 

@Setup(Level.Trial)
public void setup() {
    byte[] bytes = new byte[256];
    content = Unpooled.buffer(bytes.length);
    content.writeBytes(bytes);
    ByteBuf testContent = Unpooled.unreleasableBuffer(content.asReadOnly());

    List<RedisMessage> rList = new ArrayList<RedisMessage>(arraySize);
    for (int i = 0; i < arraySize; ++i) {
        rList.add(new FullBulkStringRedisMessage(testContent));
    }
    redisArray = new ArrayRedisMessage(rList);
    encoder = new RedisEncoder();
    context = new EmbeddedChannelWriteReleaseHandlerContext(pooledAllocator ? PooledByteBufAllocator.DEFAULT :
            UnpooledByteBufAllocator.DEFAULT, encoder) {
        @Override
        protected void handleException(Throwable t) {
            handleUnexpectedException(t);
        }
    };
}
 

@Test
default void stringify() {
    Codec<T> codec = getCodec(UnpooledByteBufAllocator.DEFAULT);
    T[] origin = originParameters();
    Object[] strings = stringifyParameters();

    assertEquals(origin.length, strings.length);

    for (int i = 0; i < origin.length; ++i) {
        ParameterWriter writer = ParameterWriterHelper.get(1);
        codec.encode(origin[i], context())
            .publishText(writer)
            .as(StepVerifier::create)
            .verifyComplete();
        assertEquals(ParameterWriterHelper.toSql(writer), strings[i].toString());
    }
}
 

public static byte[] createMessageWrittenToLedger(String msg) throws Exception {
    PulsarApi.MessageMetadata.Builder messageMetadataBuilder = PulsarApi.MessageMetadata.newBuilder();
    messageMetadataBuilder.setPublishTime(System.currentTimeMillis());
    messageMetadataBuilder.setProducerName("createMessageWrittenToLedger");
    messageMetadataBuilder.setSequenceId(1);
    PulsarApi.MessageMetadata messageMetadata = messageMetadataBuilder.build();
    ByteBuf data = UnpooledByteBufAllocator.DEFAULT.heapBuffer().writeBytes(msg.getBytes());

    int msgMetadataSize = messageMetadata.getSerializedSize();
    int payloadSize = data.readableBytes();
    int totalSize = 4 + msgMetadataSize + payloadSize;

    ByteBuf headers = PulsarByteBufAllocator.DEFAULT.heapBuffer(totalSize, totalSize);
    ByteBufCodedOutputStream outStream = ByteBufCodedOutputStream.get(headers);
    headers.writeInt(msgMetadataSize);
    messageMetadata.writeTo(outStream);
    ByteBuf headersAndPayload = ByteBufPair.coalesce(ByteBufPair.get(headers, data));
    byte[] byteMessage = headersAndPayload.nioBuffer().array();
    headersAndPayload.release();
    return byteMessage;
}
 

@Test
public void testEncryptedRecord() throws Exception {
  // If unlimited strength encryption is not available, we cant run this test.
  Assume.assumeFalse(Cipher.getMaxAllowedKeyLength("AES") < 256);

  UnpooledByteBufAllocator allocator = new UnpooledByteBufAllocator(false);
  CollectdParser parser = new CollectdParser(getContext(), false, null, false, AUTH_FILE_PATH, CHARSET);
  byte[] bytes = Files.readAllBytes(SINGLE_ENCRYPTED_PACKET.toPath());
  ByteBuf buf = allocator.buffer(bytes.length);
  buf.writeBytes(bytes);
  List<Record> records = parser.parse(buf, null, null);

  Assert.assertEquals(24, records.size()); // 24 value parts
  Record record14 = records.get(14);
  UDPTestUtil.verifyCollectdRecord(UDPTestUtil.encryptedRecord14, record14);
  LOG.info("Num records: {}", records.size());
}
 

@Test
public void testSNIMatchersDoesNotThrow() throws Exception {
    assumeTrue(PlatformDependent.javaVersion() >= 8);
    SelfSignedCertificate ssc = new SelfSignedCertificate();
    serverSslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey())
            .sslProvider(sslServerProvider())
            .build();

    SSLEngine engine = serverSslCtx.newEngine(UnpooledByteBufAllocator.DEFAULT);
    try {
        SSLParameters parameters = new SSLParameters();
        Java8SslTestUtils.setSNIMatcher(parameters);
        engine.setSSLParameters(parameters);
    } finally {
        cleanupServerSslEngine(engine);
        ssc.delete();
    }
}
 

@Test
public void usingNettyDataBufferFactory_PooledHttpData() {
    final DataBufferFactoryWrapper<?> wrapper =
            new DataBufferFactoryWrapper<>(new NettyDataBufferFactory(UnpooledByteBufAllocator.DEFAULT));

    final PooledHttpData httpData1 =
            PooledHttpData.wrap(Unpooled.wrappedBuffer("abc".getBytes()));

    final DataBuffer buffer = wrapper.toDataBuffer(httpData1);
    assertThat(buffer).isInstanceOf(NettyDataBuffer.class);
    assertThat(((NettyDataBuffer) buffer).getNativeBuffer().refCnt()).isOne();

    final HttpData httpData2 = wrapper.toHttpData(buffer);
    assertThat(httpData2).isInstanceOf(PooledHttpData.class);
    assertThat(((PooledHttpData) httpData2).content())
            .isEqualTo(((NettyDataBuffer) buffer).getNativeBuffer());
    assertThat(((PooledHttpData) httpData2).refCnt()).isOne();
}
 

private void testProtocol(String[] clientProtocols, String[] serverProtocols) throws Exception {
    SelfSignedCertificate cert = new SelfSignedCertificate();

    clientSslCtx = SslContextBuilder
            .forClient()
            .trustManager(cert.cert())
            .sslProvider(sslClientProvider())
            .protocols(clientProtocols)
            .build();
    SSLEngine client = clientSslCtx.newEngine(UnpooledByteBufAllocator.DEFAULT);

    serverSslCtx = SslContextBuilder
            .forServer(cert.certificate(), cert.privateKey())
            .sslProvider(sslServerProvider())
            .protocols(serverProtocols)
            .build();
    SSLEngine server = serverSslCtx.newEngine(UnpooledByteBufAllocator.DEFAULT);

    try {
        handshake(client, server);
    } finally {
        cleanupClientSslEngine(client);
        cleanupServerSslEngine(server);
        cert.delete();
    }
}
 

@Test
public void testBeginHandshakeAfterEngineClosed() throws SSLException {
    clientSslCtx = SslContextBuilder
            .forClient()
            .sslProvider(sslClientProvider())
            .build();
    SSLEngine client = clientSslCtx.newEngine(UnpooledByteBufAllocator.DEFAULT);

    try {
        client.closeInbound();
        client.closeOutbound();
        try {
            client.beginHandshake();
            fail();
        } catch (SSLException expected) {
            // expected
        }
    } finally {
        cleanupClientSslEngine(client);
    }
}
 

@Test
public void testBeginHandshakeCloseOutbound() throws Exception {
    SelfSignedCertificate cert = new SelfSignedCertificate();

    clientSslCtx = SslContextBuilder
            .forClient()
            .sslProvider(sslClientProvider())
            .build();
    SSLEngine client = clientSslCtx.newEngine(UnpooledByteBufAllocator.DEFAULT);

    serverSslCtx = SslContextBuilder
            .forServer(cert.certificate(), cert.privateKey())
            .sslProvider(sslServerProvider())
            .build();
    SSLEngine server = serverSslCtx.newEngine(UnpooledByteBufAllocator.DEFAULT);

    try {
        testBeginHandshakeCloseOutbound(client);
        testBeginHandshakeCloseOutbound(server);
    } finally {
        cleanupClientSslEngine(client);
        cleanupServerSslEngine(server);
        cert.delete();
    }
}
 

@Test
public void testCloseInboundAfterBeginHandshake() throws Exception {
    SelfSignedCertificate cert = new SelfSignedCertificate();

    clientSslCtx = SslContextBuilder
            .forClient()
            .sslProvider(sslClientProvider())
            .build();
    SSLEngine client = clientSslCtx.newEngine(UnpooledByteBufAllocator.DEFAULT);

    serverSslCtx = SslContextBuilder
            .forServer(cert.certificate(), cert.privateKey())
            .sslProvider(sslServerProvider())
            .build();
    SSLEngine server = serverSslCtx.newEngine(UnpooledByteBufAllocator.DEFAULT);

    try {
        testCloseInboundAfterBeginHandshake(client);
        testCloseInboundAfterBeginHandshake(server);
    } finally {
        cleanupClientSslEngine(client);
        cleanupServerSslEngine(server);
        cert.delete();
    }
}
 

@Test
public void testFullRequestWithBody() throws Exception {
  outputReceived = new CountDownLatch(1);
  ByteBuf body = ByteBufUtil.writeUtf8(UnpooledByteBufAllocator.DEFAULT, "body");
  FullRequest requestIn = RequestBuilders.newPost("/").body(body).build();

  channel.writeOutbound(requestIn);

  channel.runPendingTasks(); // blocks

  Uninterruptibles.awaitUninterruptibly(outputReceived);

  FullHttpRequest requestOut = (FullHttpRequest) requests.remove(0);

  assertTrue(requestOut != null);
  assertEquals(HTTP_1_1, requestOut.protocolVersion());
  assertEquals(HttpMethod.POST, requestOut.method());
  assertEquals("/", requestOut.uri());
  assertFalse(requestOut.content() == null);
  assertEquals(body, requestOut.content());
}
 

private Observable<Void> handleRecommendationsByUserId(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) {
    System.out.println("HTTP request -> recommendations by user id request: " + request.getPath());
    final String userId = userIdFromPath(request.getPath());
    if (userId == null) {
        response.setStatus(HttpResponseStatus.BAD_REQUEST);
        return response.close();
    }
    if (!userRecommendations.containsKey(userId)) {
        response.setStatus(HttpResponseStatus.NOT_FOUND);
        return response.close();
    }

    StringBuilder builder = new StringBuilder();
    for (String movieId : userRecommendations.get(userId)) {
        System.out.println("    returning: " + movies.get(movieId));
        builder.append(movies.get(movieId)).append('\n');
    }

    ByteBuf byteBuf = UnpooledByteBufAllocator.DEFAULT.buffer();
    byteBuf.writeBytes(builder.toString().getBytes(Charset.defaultCharset()));

    response.write(byteBuf);
    return response.close();
}
 

@Test(expected = OnRecordErrorException.class)
public void testInvalidVersion() throws Exception {
  UnpooledByteBufAllocator allocator = new UnpooledByteBufAllocator(false);
  NetflowParser netflowParser = makeNetflowParser();
  ByteBuf buf = allocator.buffer(4);
  buf.writeShort(0);
  buf.writeShort(0);
  netflowParser.parse(buf, null, null);
}
 

@Override
public IScriptBuffer protoToBuf(Builder<?> builder) {
	try{
		byte[] bytes = builder.build().toByteArray();
		ByteBuf buf = UnpooledByteBufAllocator.DEFAULT.heapBuffer(bytes.length);
		buf.writeBytes(bytes);
		IScriptBuffer bufOut = DFJsBuffer.newBuffer(buf);
		return bufOut;
	}catch(Throwable e){
		e.printStackTrace();
	}
	return null;
}
 

@Test
public void testMessageParsing() throws Exception {
  SyslogParser parser = new SyslogParser(getContext(), StandardCharsets.UTF_8);
  List<String> messages = getTestMessageStrings();

  UnpooledByteBufAllocator allocator = new UnpooledByteBufAllocator(false);
  // test with default keepFields = false
  for (String msg : messages) {
    byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
    ByteBuf buffer = allocator.buffer(bytes.length);
    buffer.writeBytes(bytes);
    List<Record> records = parser.parse(
        buffer,
        InetSocketAddress.createUnresolved("localhost", 5000),
        InetSocketAddress.createUnresolved("localhost", 50000)
    );
    Assert.assertEquals(1, records.size());
    Assert.assertEquals("Failure to parse known-good syslog message",
      msg, records.get(0).get("/raw").getValueAsString());
    Assert.assertEquals("Failure to parse known-good syslog message",
      "localhost:5000", records.get(0).get("/receiverAddr").getValueAsString());
    Assert.assertEquals("Failure to parse known-good syslog message",
      "localhost:50000", records.get(0).get("/senderAddr").getValueAsString());
    Assert.assertNotNull("Failure to parse known-good syslog message",
      records.get(0).get("/host").getValueAsString());
  }
}
 

public void verifyNoFlush(CommandEncoder commandEncoder, UnpooledByteBufAllocator allocator,
                          Object command) throws Exception {
    ChannelHandlerContext context = mock(ChannelHandlerContext.class);
    Mockito.when(context.alloc()).thenReturn(allocator);
    Mockito.when(context.executor()).thenReturn(Mockito.mock(EventExecutor.class));
    commandEncoder.write(context, command, null);
    verify(context).alloc();
    verify(context).write(Mockito.any(), Mockito.any());
    verify(context, new AtMost(10)).executor(); // Irrelevant
    verify(context, new AtMost(10)).channel(); // Irrelevant
    verifyNoMoreInteractions(context);
}
 

private Server buildChain() throws Exception {
    Server server = startEmptyServer();
    Channel channel = clientBootstrap().connect(PROXY_HOST, PROXY_PORT1).sync().channel();
    logger.info("{}", getProxyProtocol(server.getPort()));
    channel.writeAndFlush(UnpooledByteBufAllocator.DEFAULT.buffer()
            .writeBytes(getProxyProtocol(server.getPort()).getBytes()));
    Thread.sleep(1000);
    waitConditionUntilTimeOut(()->chainEstablished(), 2000);
    return server;
}
 

@Test
default void binary() {
    Codec<T> codec = getCodec(UnpooledByteBufAllocator.DEFAULT);
    T[] origin = originParameters();
    ByteBuf[] binaries = binaryParameters(CharCollation.clientCharCollation().getCharset());

    assertEquals(origin.length, binaries.length);

    for (int i = 0; i < origin.length; ++i) {
        merge(Flux.from(codec.encode(origin[i], context()).publishBinary()))
            .as(StepVerifier::create)
            .expectNext(sized(binaries[i]))
            .verifyComplete();
    }
}
 

@SuppressWarnings("unchecked")
private void writeMetric(String json, HttpServerResponse<O> response) {
    byte[] bytes = json.getBytes(Charset.defaultCharset());
    ByteBuf byteBuf = UnpooledByteBufAllocator.DEFAULT.buffer(bytes.length + EXTRA_SPACE);
    byteBuf.writeBytes(HEADER);
    byteBuf.writeBytes(bytes);
    byteBuf.writeBytes(FOOTER);
    response.writeAndFlush((O) byteBuf);
}
 

protected void setUp(final ChannelHandler encoder) throws Exception {

        channel = new EmbeddedChannel(encoder);
        channel.config().setAllocator(new UnpooledByteBufAllocator(false));
        channel.attr(ChannelAttributes.MAX_PACKET_SIZE_SEND).set((long) MAX_PACKET_SIZE);
        channel.attr(ChannelAttributes.REQUEST_PROBLEM_INFORMATION).set(true);
        channel.attr(ChannelAttributes.CLIENT_ID).set("clientId");
        channel.attr(ChannelAttributes.MQTT_VERSION).set(ProtocolVersion.MQTTv5);

    }
 

@Test
public void canMarshallFailedItemSource() throws IOException {

    // given
    CompositeByteBuf byteBuf = UnpooledByteBufAllocator.DEFAULT.compositeHeapBuffer(2);
    byte[] bytes = new byte[512];
    random.nextBytes(bytes);

    String expectedPayload = new String(bytes, Charset.defaultCharset());

    byteBuf.writeBytes(bytes);

    assertEquals(expectedPayload, byteBuf.toString(Charset.defaultCharset()));

    ItemSource<ByteBuf> expectedSource = new ByteBufItemSource(byteBuf, (source) -> byteBuf.release());

    String targetName = UUID.randomUUID().toString();
    FailedItemSource<ByteBuf> failedItemSource = new FailedItemSource<>(
            expectedSource,
            new FailedItemInfo(targetName)
    );

    ChronicleMap<CharSequence, ItemSource> map = createDefaultTestChronicleMap();
    String key = UUID.randomUUID().toString();

    // when
    map.put(key, failedItemSource);
    FailedItemSource<ByteBuf> result = (FailedItemSource<ByteBuf>) map.get(key);

    // then
    assertEquals(expectedPayload, result.getSource().toString(Charset.defaultCharset()));
    assertEquals(failedItemSource.getInfo().getTargetName(), result.getInfo().getTargetName());

}
 

/** Set up for test. */
@Before
public void setUp() {
  MockitoAnnotations.initMocks(this);

  when(channel.alloc()).thenReturn(UnpooledByteBufAllocator.DEFAULT);
  when(channel.pipeline()).thenReturn(pipeline);
  when(channel.eventLoop()).thenReturn(eventLoop);
  when(channel.newPromise()).thenReturn(new DefaultChannelPromise(channel));
  when(channel.voidPromise()).thenReturn(new DefaultChannelPromise(channel));
  ChannelPromise completedPromise = new DefaultChannelPromise(channel)
      .setSuccess();
  when(channel.write(any())).thenReturn(completedPromise);
  when(channel.writeAndFlush(any())).thenReturn(completedPromise);
  when(writeQueue.enqueue(any(QueuedCommand.class), anyBoolean())).thenReturn(completedPromise);
  when(pipeline.firstContext()).thenReturn(ctx);
  when(eventLoop.inEventLoop()).thenReturn(true);
  when(http2Stream.id()).thenReturn(STREAM_ID);

  doAnswer(new Answer<Void>() {
    @Override
    public Void answer(InvocationOnMock invocation) throws Throwable {
      Runnable runnable = (Runnable) invocation.getArguments()[0];
      runnable.run();
      return null;
    }
  }).when(eventLoop).execute(any(Runnable.class));

  stream = createStream();
}
 

@Override protected HttpClient newClient(int port) {
    return Exceptions.uncheck(() -> new ZipkinHttpClientImpl( HttpClient.of(s -> s
        .poolSize(0)
        .byteBufAllocator(UnpooledByteBufAllocator.DEFAULT)
        .maxContentLength(ServerConfig.DEFAULT_MAX_CONTENT_LENGTH)
    ), httpTracing));
}
 

@Override
public void execute(IndexTemplate indexTemplate) {

    ByteBuf byteBuf = UnpooledByteBufAllocator.DEFAULT.buffer(4096)
            .writeBytes(indexTemplate.getSource().getBytes());

    IndexTemplateRequest request = new IndexTemplateRequest.Builder()
            .withTemplateName(indexTemplate.getName())
            .withSource(byteBuf)
            .build();

    try {
        Function<Exception, BasicResponse> errorResponseTemplate =
                (ex) -> new BasicResponse().withErrorMessage("IndexTemplate not added: " + ex.getMessage());

        BlockingResponseHandler<BasicResponse> responseHandler =
                new BlockingResponseHandler<>(objectReader, errorResponseTemplate);

        Response result = createClient().execute(request, responseHandler);
        if (!result.isSucceeded()) {
            LOG.error(result.getErrorMessage());
        }
    } finally {
        byteBuf.release();
    }

}
 

@Test
public void testIndicatedSizeSmallerThanActualSizeSingleChunk() {
  HelloRequest grpcRequest = HelloRequest.newBuilder().setName("myName").build();
  byte[] dataBytes = grpcRequest.toByteArray();
  byte[] lengthByteBuffer = ByteBuffer.allocate(4).putInt(2).array();
  byte[] compressedByteBuffer = ByteBuffer.allocate(1).put((byte) 0).array();
  int streamId = 567;

  int length = dataBytes.length;
  ByteBuf grpcRequestBuffer = UnpooledByteBufAllocator.DEFAULT.buffer(length + 5, length + 5);

  grpcRequestBuffer.writeBytes(compressedByteBuffer);
  grpcRequestBuffer.writeBytes(lengthByteBuffer);
  grpcRequestBuffer.writeBytes(dataBytes);

  channel.writeInbound(fullGrpcRequest(grpcRequestBuffer, streamId, true));

  Response response = channel.readOutbound();
  SegmentedData segmentedData = channel.readOutbound();

  assertEquals(HttpResponseStatus.OK, response.status());
  assertEquals(streamId, response.streamId());
  assertEquals("application/grpc+proto", response.headers().get(HttpHeaderNames.CONTENT_TYPE));

  assertEquals("13", Objects.requireNonNull(segmentedData.trailingHeaders()).get("grpc-status"));
  String actualMessage =
      grpcDecodedString(
          Objects.requireNonNull(
              Objects.requireNonNull(segmentedData.trailingHeaders()).get("grpc-message")));
  assertEquals("indicated payload size does not match actual payload size", actualMessage);
  assertEquals(streamId, segmentedData.streamId());
  assertTrue(segmentedData.endOfMessage());
}
 

private static void testClientContext(SslProvider provider) throws Exception {
    SelfSignedCertificate cert = new SelfSignedCertificate();
    SslContextBuilder builder = SslContextBuilder.forClient()
                                                 .sslProvider(provider)
                                                 .keyManager(cert.key(), cert.cert())
                                                 .trustManager(cert.cert())
                                                 .clientAuth(ClientAuth.OPTIONAL);
    SslContext context = builder.build();
    SSLEngine engine = context.newEngine(UnpooledByteBufAllocator.DEFAULT);
    assertFalse(engine.getWantClientAuth());
    assertFalse(engine.getNeedClientAuth());
    engine.closeInbound();
    engine.closeOutbound();
}