Java 类io.netty.channel.embedded.EmbeddedChannel 实例源码

public void testThatPipeliningWorksWithFastSerializedRequests() throws InterruptedException {
    final int numberOfRequests = randomIntBetween(2, 128);
    final EmbeddedChannel embeddedChannel = new EmbeddedChannel(new HttpPipeliningHandler(numberOfRequests), new WorkEmulatorHandler());

    for (int i = 0; i < numberOfRequests; i++) {
        embeddedChannel.writeInbound(createHttpRequest("/" + String.valueOf(i)));
    }

    final List<CountDownLatch> latches = new ArrayList<>();
    for (final String url : waitingRequests.keySet()) {
        latches.add(finishRequest(url));
    }

    for (final CountDownLatch latch : latches) {
        latch.await();
    }

    embeddedChannel.flush();

    for (int i = 0; i < numberOfRequests; i++) {
        assertReadHttpMessageHasContent(embeddedChannel, String.valueOf(i));
    }

    assertTrue(embeddedChannel.isOpen());
}

public void testThatPipeliningClosesConnectionWithTooManyEvents() throws InterruptedException {
    final int numberOfRequests = randomIntBetween(2, 128);
    final EmbeddedChannel embeddedChannel = new EmbeddedChannel(new HttpPipeliningHandler(numberOfRequests), new WorkEmulatorHandler());

    for (int i = 0; i < 1 + numberOfRequests + 1; i++) {
        embeddedChannel.writeInbound(createHttpRequest("/" + Integer.toString(i)));
    }

    final List<CountDownLatch> latches = new ArrayList<>();
    final List<Integer> requests = IntStream.range(1, numberOfRequests + 1).mapToObj(r -> r).collect(Collectors.toList());
    Randomness.shuffle(requests);

    for (final Integer request : requests) {
        latches.add(finishRequest(request.toString()));
    }

    for (final CountDownLatch latch : latches) {
        latch.await();
    }

    finishRequest(Integer.toString(numberOfRequests + 1)).await();

    embeddedChannel.flush();

    assertFalse(embeddedChannel.isOpen());
}

public void testReleaseOnSendToClosedChannel() {
    final Settings settings = Settings.builder().build();
    final NamedXContentRegistry registry = xContentRegistry();
    try (Netty4HttpServerTransport httpServerTransport =
                 new Netty4HttpServerTransport(settings, networkService, bigArrays, threadPool, registry, new NullDispatcher())) {
        final FullHttpRequest httpRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/");
        final EmbeddedChannel embeddedChannel = new EmbeddedChannel();
        final Netty4HttpRequest request = new Netty4HttpRequest(registry, httpRequest, embeddedChannel);
        final HttpPipelinedRequest pipelinedRequest = randomBoolean() ? new HttpPipelinedRequest(request.request(), 1) : null;
        final Netty4HttpChannel channel =
                new Netty4HttpChannel(httpServerTransport, request, pipelinedRequest, randomBoolean(), threadPool.getThreadContext());
        final TestResponse response = new TestResponse(bigArrays);
        assertThat(response.content(), instanceOf(Releasable.class));
        embeddedChannel.close();
        channel.sendResponse(response);
        // ESTestCase#after will invoke ensureAllArraysAreReleased which will fail if the response content was not released
    }
}

@Test
@Tag("fast")
public void encode() {
    IsoOnTcpMessage isoOnTcpMessage = new IsoOnTcpMessage(
        Unpooled.wrappedBuffer(new byte[]{(byte)0x01,(byte)0x02,(byte)0x03}));
    EmbeddedChannel channel = new EmbeddedChannel(new IsoOnTcpProtocol());
    channel.writeOutbound(isoOnTcpMessage);
    channel.checkException();
    Object obj = channel.readOutbound();
    assertThat(obj).isInstanceOf(ByteBuf.class);
    ByteBuf byteBuf = (ByteBuf) obj;
    assertEquals(4 + 3, byteBuf.readableBytes(),
        "The TCP on ISO Header should add 4 bytes to the data sent");
    assertEquals(IsoOnTcpProtocol.ISO_ON_TCP_MAGIC_NUMBER, byteBuf.getByte(0));
    assertEquals(4 + 3, byteBuf.getShort(2),
        "The length value in the packet should reflect the size of the entire data being sent");
}

/**
 * Happy path test.
 */
@Test
@Tag("fast")
public void decode() {
    EmbeddedChannel channel = new EmbeddedChannel(new IsoOnTcpProtocol());
    channel.writeInbound(Unpooled.wrappedBuffer(new byte[]{IsoOnTcpProtocol.ISO_ON_TCP_MAGIC_NUMBER,
        (byte)0x00,(byte)0x00,(byte)0x0D,
        (byte)0x01,(byte)0x02,(byte)0x03,(byte)0x04,(byte)0x05,(byte)0x06,(byte)0x07,(byte)0x08,(byte)0x09}));
    channel.checkException();
    Object obj = channel.readInbound();
    assertThat(obj).isInstanceOf(IsoOnTcpMessage.class);
    IsoOnTcpMessage isoOnTcpMessage = (IsoOnTcpMessage) obj;
    assertNotNull(isoOnTcpMessage.getUserData());
    assertEquals(9, isoOnTcpMessage.getUserData().readableBytes());
}

/**
 * Passes this Packet on to the NetHandler for processing.
 */
@Override
public void processPacket(INetHandler inethandler)
{
    this.netHandler = inethandler;
    EmbeddedChannel internalChannel = NetworkRegistry.INSTANCE.getChannel(this.channel, this.target);
    if (internalChannel != null)
    {
        internalChannel.attr(NetworkRegistry.NET_HANDLER).set(this.netHandler);
        try
        {
            if (internalChannel.writeInbound(this))
            {
                badPackets.add(this.channel);
                if (badPackets.size() % packetCountWarning == 0)
                {
                    FMLLog.severe("Detected ongoing potential memory leak. %d packets have leaked. Top offenders", badPackets.size());
                    int i = 0;
                    for (Entry<String> s  : Multisets.copyHighestCountFirst(badPackets).entrySet())
                    {
                        if (i++ > 10) break;
                        FMLLog.severe("\t %s : %d", s.getElement(), s.getCount());
                    }
                }
            }
            internalChannel.inboundMessages().clear();
        }
        catch (FMLNetworkException ne)
        {
            FMLLog.log(Level.ERROR, ne, "There was a network exception handling a packet on channel %s", channel);
            dispatcher.rejectHandshake(ne.getMessage());
        }
        catch (Throwable t)
        {
            FMLLog.log(Level.ERROR, t, "There was a critical exception handling a packet on channel %s", channel);
            dispatcher.rejectHandshake("A fatal error has occurred, this connection is terminated");
        }
    }
}

private MqttConnAckMessage executeNormalChannelRead0(String clientId, boolean cleanSession, ChannelId channelId)
        throws Exception {
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false,
            10);
    MqttConnectVariableHeader variableHeader = new MqttConnectVariableHeader("MQTT", 4, true, true, true, 0, true,
            cleanSession, 60);
    MqttConnectPayload payload = new MqttConnectPayload(clientId, "willtopic", "willmessage", "username",
            "password");

    MqttConnectMessage msg = new MqttConnectMessage(fixedHeader, variableHeader, payload);

    ChannelId cid = channelId == null ? TestUtil.newChannelId(clientId, false) : channelId;

    EmbeddedChannel channel = new EmbeddedChannel(cid, new ConnectReceiver());

    channel.writeInbound(msg);

    return channel.readOutbound();
}

@Before
public void prepare () throws PropertyVetoException, SQLException {
    Security.addProvider(new org.bouncycastle.jce.provider.BouncyCastleProvider());

    node1.isServer = false;
    node2.isServer = true;

    contextFactory1 = new MockContextFactory(serverObject1);
    contextFactory2 = new MockContextFactory(serverObject2);

    node1.ephemeralKeyClient = node2.ephemeralKeyServer;
    node2.ephemeralKeyClient = node1.ephemeralKeyServer;

    node1.ecdhKeySet = ECDH.getSharedSecret(node1.ephemeralKeyServer, node1.ephemeralKeyClient);
    node2.ecdhKeySet = ECDH.getSharedSecret(node2.ephemeralKeyServer, node2.ephemeralKeyClient);

    channel1 = new EmbeddedChannel(new ProcessorHandler(contextFactory1.getAuthenticationProcessor(node1), "Encryption1"));
    channel2 = new EmbeddedChannel(new ProcessorHandler(contextFactory2.getAuthenticationProcessor(node2), "Encryption2"));

    Message m = (Message) channel2.readOutbound();
    assertNull(m);

}

@Before
public void prepare () {
    node1.isServer = false;
    node1.intent = ConnectionIntent.GET_IPS;

    node2.isServer = true;

    contextFactory1 = new MockContextFactory(serverObject1, dbHandler1);
    contextFactory2 = new MockContextFactory(serverObject2, dbHandler2);

    dbHandler2.fillWithRandomData();

    channel1 = new EmbeddedChannel(new ProcessorHandler(contextFactory1.getPeerSeedProcessor(node1), "Seed1"));
    channel2 = new EmbeddedChannel(new ProcessorHandler(contextFactory2.getPeerSeedProcessor(node2), "Seed2"));

    Message m = (Message) channel2.readOutbound();
    assertNull(m);
}

public void prepare () {
    node1 = new ClientObject();
    node2 = new ClientObject();

    node1.isServer = false;
    node2.isServer = true;

    contextFactory1 = new MockContextFactory(serverObject1, dbHandler1);
    contextFactory2 = new MockContextFactory(serverObject2, dbHandler2);

    channel1 = new EmbeddedChannel(new ProcessorHandler(contextFactory1.getSyncProcessor(node1), "Sync1"));
    channel2 = new EmbeddedChannel(new ProcessorHandler(contextFactory2.getSyncProcessor(node2), "Sync2"));

    Message m = (Message) channel2.readOutbound();
    assertNull(m);
}

@Before
public void prepare () throws PropertyVetoException, SQLException {
    node12.name = "LNPayment12";
    node21.name = "LNPayment21";
    node23.name = "LNPayment23";
    node32.name = "LNPayment32";

    node12.pubKeyClient = node2.pubKeyServer;
    node21.pubKeyClient = node1.pubKeyServer;
    node23.pubKeyClient = node3.pubKeyServer;
    node32.pubKeyClient = node2.pubKeyServer;

    processor12 = new LNPaymentProcessorImpl(contextFactory1, dbHandler1, node12);
    processor21 = new LNPaymentProcessorImpl(contextFactory2, dbHandler2, node21);
    processor23 = new LNPaymentProcessorImpl(contextFactory2, dbHandler2, node23);
    processor32 = new LNPaymentProcessorImpl(contextFactory3, dbHandler3, node32);

    channel12 = new EmbeddedChannel(new ProcessorHandler(processor12, "LNPayment12"));
    channel21 = new EmbeddedChannel(new ProcessorHandler(processor21, "LNPayment21"));
    channel23 = new EmbeddedChannel(new ProcessorHandler(processor23, "LNPayment23"));
    channel32 = new EmbeddedChannel(new ProcessorHandler(processor32, "LNPayment32"));

    Message m = (Message) channel21.readOutbound();
    assertNull(m);

}

@Before
public void prepare () throws PropertyVetoException, SQLException {
    node1.isServer = false;
    node1.intent = ConnectionIntent.OPEN_CHANNEL;
    node2.isServer = true;

    contextFactory1 = new EstablishMockContextFactory(serverObject1, dbHandler1);
    contextFactory2 = new EstablishMockContextFactory(serverObject2, dbHandler2);

    processor1 = new LNEstablishProcessorImpl(contextFactory1, dbHandler1, node1);
    processor2 = new LNEstablishProcessorImpl(contextFactory2, dbHandler2, node2);

    channel1 = new EmbeddedChannel(new ProcessorHandler(processor1, "LNEstablish1"));
    channel2 = new EmbeddedChannel(new ProcessorHandler(processor2, "LNEstablish2"));

    contextFactory1.getChannelManager().openChannel(new NodeKey(node1.pubKeyClient), new ChannelOpenListener());

    Message m = (Message) channel2.readOutbound();
    assertNull(m);
}

@Before
public void prepare () throws PropertyVetoException, SQLException {
    node1.isServer = false;
    node2.isServer = true;

    this.node1.name = "LNPayment12";
    this.node2.name = "LNPayment21";

    processor12 = new LNPaymentProcessorImpl(contextFactory12, dbHandler1, this.node1);
    processor21 = new LNPaymentProcessorImpl(contextFactory21, dbHandler2, this.node2);

    channel12 = new EmbeddedChannel(new ProcessorHandler(processor12, "LNPayment12"));
    channel21 = new EmbeddedChannel(new ProcessorHandler(processor21, "LNPayment21"));

    Message m = (Message) channel21.readOutbound();
    assertNull(m);

}

public void connectChannel (EmbeddedChannel from, EmbeddedChannel to) {
    new Thread(new Runnable() {
        @Override
        public void run () {
            while (true) {
                TestTools.exchangeMessagesDuplex(from, to);
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        }
    }).start();
}

@Test
public void addProxyHandler() {
    ClientOptions.Builder<?> builder = ClientOptions.builder();
    EmbeddedChannel channel = new EmbeddedChannel();

    ClientContextHandler.addProxyHandler(builder.build(), channel.pipeline(),
            new InetSocketAddress("localhost", 8080));
    assertThat(channel.pipeline().get(NettyPipeline.ProxyHandler)).isNull();

    builder.proxy(ops -> ops.type(Proxy.HTTP)
                            .host("proxy")
                            .port(8080));
    ClientContextHandler.addProxyHandler(builder.build(), channel.pipeline(),
            new InetSocketAddress("localhost", 8080));
    assertThat(channel.pipeline().get(NettyPipeline.ProxyHandler)).isNull();
}

@Test
public void onWriteIdleReplaces() throws Exception {
    EmbeddedChannel channel = new EmbeddedChannel();
    NettyContext mockContext = () -> channel;
    NettyOutbound outbound = () -> mockContext;

    AtomicLong idle1 = new AtomicLong();
    AtomicLong idle2 = new AtomicLong();

    outbound.onWriteIdle(100, idle1::incrementAndGet);
    outbound.onWriteIdle(150, idle2::incrementAndGet);
    ReactorNetty.OutboundIdleStateHandler idleStateHandler =
            (ReactorNetty.OutboundIdleStateHandler) channel.pipeline().get(NettyPipeline.OnChannelWriteIdle);
    idleStateHandler.onWriteIdle.run();

    assertThat(channel.pipeline().names()).containsExactly(
            NettyPipeline.OnChannelWriteIdle,
            "DefaultChannelPipeline$TailContext#0");

    assertThat(idle1.intValue()).isZero();
    assertThat(idle2.intValue()).isEqualTo(1);
}

@Test
public void addDecoderReplaysLastHttp() throws Exception {
    ByteBuf buf = Unpooled.copiedBuffer("{\"foo\":1}", CharsetUtil.UTF_8);
    EmbeddedChannel channel = new EmbeddedChannel();
    HttpClientOperations ops = new HttpClientOperations(channel,
            (response, request) -> null, handler);

    ops.addHandler(new JsonObjectDecoder());
    channel.writeInbound(new DefaultLastHttpContent(buf));

    assertThat(channel.pipeline().names().iterator().next(), is("JsonObjectDecoder$extractor"));

    Object content = channel.readInbound();
    assertThat(content, instanceOf(ByteBuf.class));
    ((ByteBuf) content).release();

    content = channel.readInbound();
    assertThat(content, instanceOf(LastHttpContent.class));
    ((LastHttpContent) content).release();

    assertThat(channel.readInbound(), nullValue());
}

@Test
public void addNamedDecoderReplaysLastHttp() throws Exception {
    ByteBuf buf = Unpooled.copiedBuffer("{\"foo\":1}", CharsetUtil.UTF_8);
    EmbeddedChannel channel = new EmbeddedChannel();
    HttpClientOperations ops = new HttpClientOperations(channel,
            (response, request) -> null, handler);

    ops.addHandler("json", new JsonObjectDecoder());
    channel.writeInbound(new DefaultLastHttpContent(buf));

    assertThat(channel.pipeline().names().iterator().next(), is("json$extractor"));

    Object content = channel.readInbound();
    assertThat(content, instanceOf(ByteBuf.class));
    ((ByteBuf) content).release();

    content = channel.readInbound();
    assertThat(content, instanceOf(LastHttpContent.class));
    ((LastHttpContent) content).release();

    assertThat(channel.readInbound(), nullValue());
}

@Test
public void addEncoderReplaysLastHttp() throws Exception {
    ByteBuf buf = Unpooled.copiedBuffer("{\"foo\":1}", CharsetUtil.UTF_8);
    EmbeddedChannel channel = new EmbeddedChannel();
    HttpClientOperations ops = new HttpClientOperations(channel,
            (response, request) -> null, handler);

    ops.addHandler(new JsonObjectDecoder());
    channel.writeInbound(new DefaultLastHttpContent(buf));

    assertThat(channel.pipeline().names().iterator().next(), is("JsonObjectDecoder$extractor"));

    Object content = channel.readInbound();
    assertThat(content, instanceOf(ByteBuf.class));
    ((ByteBuf) content).release();

    content = channel.readInbound();
    assertThat(content, instanceOf(LastHttpContent.class));
    ((LastHttpContent) content).release();

    assertThat(channel.readInbound(), nullValue());
}

@Test
public void addNamedEncoderReplaysLastHttp() throws Exception {
    ByteBuf buf = Unpooled.copiedBuffer("{\"foo\":1}", CharsetUtil.UTF_8);
    EmbeddedChannel channel = new EmbeddedChannel();
    HttpClientOperations ops = new HttpClientOperations(channel,
            (response, request) -> null, handler);

    ops.addHandler("json", new JsonObjectDecoder());
    channel.writeInbound(new DefaultLastHttpContent(buf));

    assertThat(channel.pipeline().names().iterator().next(), is("json$extractor"));

    Object content = channel.readInbound();
    assertThat(content, instanceOf(ByteBuf.class));
    ((ByteBuf) content).release();

    content = channel.readInbound();
    assertThat(content, instanceOf(LastHttpContent.class));
    ((LastHttpContent) content).release();

    assertThat(channel.readInbound(), nullValue());
}

@Test
public void testConstructorWithProvidedReplacement() {
    EmbeddedChannel channel = new EmbeddedChannel();
    channel.pipeline().addFirst(NettyPipeline.SslHandler, new ChannelHandlerAdapter() {
    });

    HttpClientOperations ops1 = new HttpClientOperations(channel,
            (response, request) -> null, handler);
    ops1.followRedirect();
    ops1.failOnClientError(false);
    ops1.failOnServerError(false);

    HttpClientOperations ops2 = new HttpClientOperations(channel, ops1);

    assertSame(ops1.channel(), ops2.channel());
    assertSame(ops1.started, ops2.started);
    assertSame(ops1.redirectedFrom, ops2.redirectedFrom);
    assertSame(ops1.isSecure, ops2.isSecure);
    assertSame(ops1.nettyRequest, ops2.nettyRequest);
    assertSame(ops1.responseState, ops2.responseState);
    assertSame(ops1.redirectable, ops2.redirectable);
    assertSame(ops1.inboundPrefetch, ops2.inboundPrefetch);
    assertSame(ops1.requestHeaders, ops2.requestHeaders);
    assertSame(ops1.clientError, ops2.clientError);
    assertSame(ops1.serverError, ops2.serverError);
}

@Test
public void sendReceivingAppend() throws Exception {
    String segment = "123";
    ByteBuf data = Unpooled.wrappedBuffer("Hello world\n".getBytes());
    StreamSegmentStore store = this.serviceBuilder.createStreamSegmentService();

    @Cleanup
    EmbeddedChannel channel = createChannel(store);

    SegmentCreated created = (SegmentCreated) sendRequest(channel, new CreateSegment(1, segment, CreateSegment.NO_SCALE, 0));
    assertEquals(segment, created.getSegment());

    UUID uuid = UUID.randomUUID();
    AppendSetup setup = (AppendSetup) sendRequest(channel, new SetupAppend(2, uuid, segment));

    assertEquals(segment, setup.getSegment());
    assertEquals(uuid, setup.getWriterId());

    DataAppended ack = (DataAppended) sendRequest(channel,
                                                  new Append(segment, uuid, data.readableBytes(), data, null));
    assertEquals(uuid, ack.getWriterId());
    assertEquals(data.readableBytes(), ack.getEventNumber());
    assertEquals(Long.MIN_VALUE, ack.getPreviousEventNumber());
}

static Reply sendRequest(EmbeddedChannel channel, Request request) throws Exception {
    channel.writeInbound(request);
    Object encodedReply = channel.readOutbound();
    for (int i = 0; encodedReply == null && i < 50; i++) {
        channel.runPendingTasks();
        Thread.sleep(10);
        encodedReply = channel.readOutbound();
    }
    if (encodedReply == null) {
        throw new IllegalStateException("No reply to request: " + request);
    }
    WireCommand decoded = CommandDecoder.parseCommand((ByteBuf) encodedReply);
    ((ByteBuf) encodedReply).release();
    assertNotNull(decoded);
    return (Reply) decoded;
}

/**
 * This unit test case ensures that {@link FileRegionEncoder} indeed wraps {@link FileRegion} to
 * {@link ByteBuf}.
 * @throws IOException if there is an error.
 */
@Test
public void testEncode() throws IOException {
    FileRegionEncoder fileRegionEncoder = new FileRegionEncoder();
    EmbeddedChannel channel = new EmbeddedChannel(fileRegionEncoder);
    File file = File.createTempFile(UUID.randomUUID().toString(), ".data");
    file.deleteOnExit();
    Random random = new Random(System.currentTimeMillis());
    int dataLength = 1 << 10;
    byte[] data = new byte[dataLength];
    random.nextBytes(data);
    write(file, data);
    FileRegion fileRegion = new DefaultFileRegion(file, 0, dataLength);
    Assert.assertEquals(0, fileRegion.transfered());
    Assert.assertEquals(dataLength, fileRegion.count());
    Assert.assertTrue(channel.writeOutbound(fileRegion));
    ByteBuf out = (ByteBuf) channel.readOutbound();
    byte[] arr = new byte[out.readableBytes()];
    out.getBytes(0, arr);
    Assert.assertArrayEquals("Data should be identical", data, arr);
}

@Test
public void testSuccess_ConnectionMetrics_twoConnections() {
  assertThat(channel.isActive()).isTrue();
  verify(metrics).registerActiveConnection(PROTOCOL, CLIENT_HASH, channel);

  // Setup second channel.
  WhoisServiceHandler whoisServiceHandler2 =
      new WhoisServiceHandler(RELAY_HOST, RELAY_PATH, () -> ACCESS_TOKEN, metrics);
  EmbeddedChannel channel2 =
      // We need a new channel id so that it has a different hash code.
      // This only is needed for EmbeddedChannel because it has a dummy hash code implementation.
      new EmbeddedChannel(DefaultChannelId.newInstance(), whoisServiceHandler2);
  assertThat(channel2.isActive()).isTrue();
  verify(metrics).registerActiveConnection(PROTOCOL, CLIENT_HASH, channel2);
  verifyNoMoreInteractions(metrics);
}

@Test
public void testSuccess_connectionMetrics_twoConnections_sameClient() throws Exception {
  setHandshakeSuccess();
  String certHash = getCertificateHash(clientCertificate);
  assertThat(channel.isActive()).isTrue();

  // Setup the second channel.
  EppServiceHandler eppServiceHandler2 =
      new EppServiceHandler(
          RELAY_HOST,
          RELAY_PATH,
          () -> ACCESS_TOKEN,
          SERVER_HOSTNAME,
          HELLO.getBytes(UTF_8),
          metrics);
  EmbeddedChannel channel2 = setUpNewChannel(eppServiceHandler2);
  setHandshakeSuccess(channel2, clientCertificate);

  assertThat(channel2.isActive()).isTrue();

  verify(metrics).registerActiveConnection(PROTOCOL, certHash, channel);
  verify(metrics).registerActiveConnection(PROTOCOL, certHash, channel2);
  verifyNoMoreInteractions(metrics);
}

@Test
public void testSuccess_connectionMetrics_twoConnections_differentClients() throws Exception {
  setHandshakeSuccess();
  String certHash = getCertificateHash(clientCertificate);
  assertThat(channel.isActive()).isTrue();

  // Setup the second channel.
  EppServiceHandler eppServiceHandler2 =
      new EppServiceHandler(
          RELAY_HOST,
          RELAY_PATH,
          () -> ACCESS_TOKEN,
          SERVER_HOSTNAME,
          HELLO.getBytes(UTF_8),
          metrics);
  EmbeddedChannel channel2 = setUpNewChannel(eppServiceHandler2);
  X509Certificate clientCertificate2 = new SelfSignedCertificate().cert();
  setHandshakeSuccess(channel2, clientCertificate2);
  String certHash2 = getCertificateHash(clientCertificate2);

  assertThat(channel2.isActive()).isTrue();

  verify(metrics).registerActiveConnection(PROTOCOL, certHash, channel);
  verify(metrics).registerActiveConnection(PROTOCOL, certHash2, channel2);
  verifyNoMoreInteractions(metrics);
}

@Before
public void setUp() {
  EmbeddedChannel frontendChannel = new EmbeddedChannel();
  frontendChannel.attr(PROTOCOL_KEY).set(frontendProtocol);
  frontendChannel.attr(CLIENT_CERTIFICATE_HASH_KEY).set(CLIENT_CERT_HASH);
  channel =
      new EmbeddedChannel(
          new ChannelInitializer<EmbeddedChannel>() {
            @Override
            protected void initChannel(EmbeddedChannel ch) throws Exception {
              ch.attr(PROTOCOL_KEY).set(backendProtocol);
              ch.attr(RELAY_CHANNEL_KEY).set(frontendChannel);
              ch.pipeline().addLast(handler);
            }
          });
}

@Test
public void testSuccess_oneConnection() {
  EmbeddedChannel channel = new EmbeddedChannel();
  metrics.registerActiveConnection(PROTOCOL, CERT_HASH, channel);
  assertThat(channel.isActive()).isTrue();
  assertThat(FrontendMetrics.activeConnectionsGauge)
      .hasValueForLabels(1, PROTOCOL, CERT_HASH)
      .and()
      .hasNoOtherValues();
  assertThat(FrontendMetrics.totalConnectionsCounter)
      .hasValueForLabels(1, PROTOCOL, CERT_HASH)
      .and()
      .hasNoOtherValues();

  ChannelFuture unusedFuture = channel.close();
  assertThat(channel.isActive()).isFalse();
  assertThat(FrontendMetrics.activeConnectionsGauge).hasNoOtherValues();
  assertThat(FrontendMetrics.totalConnectionsCounter)
      .hasValueForLabels(1, PROTOCOL, CERT_HASH)
      .and()
      .hasNoOtherValues();
}

@Test
public void testDecoder() {
    ByteBuf buf = Unpooled.buffer();
    buf.writeBytes(VALUE.getBytes());
    ByteBuf input = buf.duplicate();
    AddressedEnvelope<Object, InetSocketAddress> addressedEnvelop =
            new DefaultAddressedEnvelope<Object, InetSocketAddress>(input, new InetSocketAddress(8888));
    EmbeddedChannel channel = new EmbeddedChannel(ChannelHandlerFactories.newByteArrayDecoder("udp").newChannelHandler());
    Assert.assertTrue(channel.writeInbound(addressedEnvelop));
    Assert.assertTrue(channel.finish());
    AddressedEnvelope<Object, InetSocketAddress> result = (AddressedEnvelope) channel.readInbound();
    Assert.assertEquals(result.recipient().getPort(), addressedEnvelop.recipient().getPort());
    Assert.assertTrue(result.content() instanceof byte[]);
    Assert.assertEquals(VALUE, new String((byte[]) result.content()));
    Assert.assertNull(channel.readInbound());
}

@Test
public void v1Test() {
    EmbeddedChannel channel = new EmbeddedChannel(new VotifierProtocolDifferentiator(true, true));

    VotifierSession session = new VotifierSession();
    channel.attr(VotifierSession.KEY).set(session);

    ByteBuf test = Unpooled.buffer(256);
    for (int i = 0; i < 256; i++) {
        test.writeByte(0);
    }
    channel.writeInbound(test);

    assertEquals(VotifierSession.ProtocolVersion.ONE, session.getVersion());
    channel.close();
}

@Test(expected = DecoderException.class)
public void failIfv1NotSupported() {
    EmbeddedChannel channel = new EmbeddedChannel(new VotifierProtocolDifferentiator(true, false));

    VotifierSession session = new VotifierSession();
    channel.attr(VotifierSession.KEY).set(session);

    ByteBuf test = Unpooled.buffer(256);
    for (int i = 0; i < 256; i++) {
        test.writeByte(0);
    }
    channel.writeInbound(test);

    assertEquals(VotifierSession.ProtocolVersion.ONE, session.getVersion());
    channel.close();
}

@Test
public void testFailFastTooLongFrameRecovery() throws Exception {
    EmbeddedChannel ch = new EmbeddedChannel(
            new LengthFieldBasedFrameDecoder(5, 0, 4, 0, 4));

    for (int i = 0; i < 2; i ++) {
        try {
            assertTrue(ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0, 0, 2 })));
            fail(DecoderException.class.getSimpleName() + " must be raised.");
        } catch (TooLongFrameException e) {
            // Expected
        }

        ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0, 0, 0, 0, 1, 'A' }));
        ByteBuf buf = releaseLater((ByteBuf) ch.readInbound());
        assertEquals("A", buf.toString(CharsetUtil.ISO_8859_1));
        buf.release();
    }
}

@Test(expected = DecoderException.class)
public void testFailureDecodeBadPacket() throws Exception {
    // Create a well-formed request
    EmbeddedChannel channel = createChannel();

    Vote vote = new Vote("Test", "test", "test", "0");
    JSONObject object = new JSONObject();
    JSONObject payload = vote.serialize();
    payload.put("challenge", SESSION.getChallenge());
    object.put("payload", payload.toString());
    // We "forget" the signature.

    try {
        channel.writeInbound(object.toString());
    } finally {
        channel.close();
    }
}

@Test(expected = DecoderException.class)
public void testFailureDecodeBadVoteField() throws Exception {
    // Create a well-formed request
    EmbeddedChannel channel = createChannel();

    Vote vote = new Vote("Test", "test", "test", "0");
    JSONObject object = new JSONObject();
    JSONObject payload = vote.serialize();
    // We "forget" the challenge.
    object.put("payload", payload.toString());
    Mac mac = Mac.getInstance("HmacSHA256");
    mac.init(TestVotifierPlugin.getI().getTokens().get("default"));
    object.put("signature",
            Base64.getEncoder().encodeToString(mac.doFinal(payload.toString().getBytes(StandardCharsets.UTF_8))));

    try {
        channel.writeInbound(object.toString());
    } finally {
        channel.close();
    }
}

@Test
public void testEncoder() {
    String writeData = "안녕하세요";
    ByteBuf request = Unpooled.wrappedBuffer(writeData.getBytes());

    Base64Encoder encoder = new Base64Encoder();
    EmbeddedChannel embeddedChannel = new EmbeddedChannel(encoder);

    embeddedChannel.writeOutbound(request);
    ByteBuf response = (ByteBuf) embeddedChannel.readOutbound();

    String expect = "7JWI64WV7ZWY7IS47JqU";
    assertEquals(expect, response.toString(Charset.defaultCharset()));

    embeddedChannel.finish();
}

@Test
public void shouldInterceptPing() throws Exception {

    EmbeddedChannel channel = new EmbeddedChannel(
            new PingHandler()
    );

    TFrame frame = new TFrame(0, FrameType.PingRequest.byteValue(), Integer.MAX_VALUE, Unpooled.EMPTY_BUFFER);

    channel.writeInbound(MessageCodec.decode(frame));
    TFrame newFrame = channel.readOutbound();

    assertNotNull(newFrame);
    assertEquals(frame.size, newFrame.size);
    assertEquals(FrameType.PingResponse.byteValue(), newFrame.type);
    assertEquals(frame.id, newFrame.id);

}

@Test
public void testInvalidCallBeforeInitRequest() throws Exception {
    // Given
    EmbeddedChannel channel = new EmbeddedChannel(
            new InitRequestHandler(new PeerManager(new Bootstrap()))
    );

    CallRequestFrame callRequestFrame = Fixtures.callRequest(0, false, Unpooled.EMPTY_BUFFER);

    channel.writeInbound(
        MessageCodec.encode(
            MessageCodec.encode(PooledByteBufAllocator.DEFAULT, callRequestFrame)
        )
    );

    TFrame tFrame = MessageCodec.decode((ByteBuf) channel.readOutbound());
    ErrorFrame errorFrame = (ErrorFrame) MessageCodec.decode(tFrame);
    tFrame.release();
    assertNotNull(errorFrame);
    assertThat(errorFrame.getErrorType(), is(ErrorType.FatalProtocolError));
    assertThat(errorFrame.getMessage(), containsString("The first frame should be an Init Request"));

    channel.writeOutbound();
}

private static void assertWriteFails(ChannelHandler handler, int count) {
    final ByteBuf buffer = Unpooled.copiedBuffer("Test", CharsetUtil.US_ASCII);
    final EmbeddedChannel channel = new EmbeddedChannel(handler);
    ByteBuf[] buffers = new ByteBuf[count];
    for (int i = 0; i < buffers.length; i++) {
        buffers[i] = buffer.duplicate().retain();
    }
    try {
        assertFalse(channel.writeOutbound(buffers));
        fail();
    } catch (Exception e) {
        assertTrue(e instanceof TestException);
    }
    assertFalse(channel.finish());
    channel.closeFuture().syncUninterruptibly();

    buffer.release();
    assertNull(channel.readOutbound());
}

@Test
public void testExceedMaxNumConnections() {
    ConnectionLimitingHandler handler = new ConnectionLimitingHandler(1);

    EmbeddedChannel ch1 = new EmbeddedChannel(handler);
    ch1.writeInbound(ch1);
    assertThat(handler.numConnections()).isEqualTo(1);
    assertThat(ch1.isActive()).isTrue();

    EmbeddedChannel ch2 = new EmbeddedChannel(handler);
    ch2.writeInbound(ch2);
    assertThat(handler.numConnections()).isEqualTo(1);
    assertThat(ch2.isActive()).isFalse();

    ch1.close();
    assertThat(handler.numConnections()).isEqualTo(0);
}