MicrometerDuplexConnection implements DuplexConnection { @Override public void sendFrame(int streamId, ByteBuf frame) { frameCounters.accept(frame); delegate.sendFrame(streamId, frame); } MicrometerDuplexConnection(
      Type connectionType, DuplexConnection delegate, MeterRegistry meterRegistry, Tag... tags); @Override ByteBufAllocator alloc(); @Override SocketAddress remoteAddress(); @Override void dispose(); @Override Mono<Void> onClose(); @Override Flux<ByteBuf> receive(); @Override void sendFrame(int streamId, ByteBuf frame); @Override void sendErrorAndClose(RSocketErrorException e);  }
@DisplayName("send gathers metrics") @SuppressWarnings("unchecked") @Test void send() { ArgumentCaptor<ByteBuf> captor = ArgumentCaptor.forClass(ByteBuf.class); doNothing().when(delegate).sendFrame(Mockito.anyInt(), captor.capture()); final MicrometerDuplexConnection micrometerDuplexConnection = new MicrometerDuplexConnection( SERVER, delegate, meterRegistry, Tag.of("test-key", "test-value")); micrometerDuplexConnection.sendFrame(1, createTestCancelFrame()); micrometerDuplexConnection.sendFrame(1, createTestErrorFrame()); micrometerDuplexConnection.sendFrame(1, createTestKeepaliveFrame()); micrometerDuplexConnection.sendFrame(1, createTestLeaseFrame()); micrometerDuplexConnection.sendFrame(1, createTestMetadataPushFrame()); micrometerDuplexConnection.sendFrame(1, createTestPayloadFrame()); micrometerDuplexConnection.sendFrame(1, createTestRequestChannelFrame()); micrometerDuplexConnection.sendFrame(1, createTestRequestFireAndForgetFrame()); micrometerDuplexConnection.sendFrame(1, createTestRequestNFrame()); micrometerDuplexConnection.sendFrame(1, createTestRequestResponseFrame()); micrometerDuplexConnection.sendFrame(1, createTestRequestStreamFrame()); micrometerDuplexConnection.sendFrame(1, createTestSetupFrame()); StepVerifier.create(Flux.fromIterable(captor.getAllValues())) .expectNextCount(12) .verifyComplete(); assertThat(findCounter(SERVER, CANCEL).count()).isEqualTo(1); assertThat(findCounter(SERVER, COMPLETE).count()).isEqualTo(1); assertThat(findCounter(SERVER, ERROR).count()).isEqualTo(1); assertThat(findCounter(SERVER, KEEPALIVE).count()).isEqualTo(1); assertThat(findCounter(SERVER, LEASE).count()).isEqualTo(1); assertThat(findCounter(SERVER, METADATA_PUSH).count()).isEqualTo(1); assertThat(findCounter(SERVER, REQUEST_CHANNEL).count()).isEqualTo(1); assertThat(findCounter(SERVER, REQUEST_FNF).count()).isEqualTo(1); assertThat(findCounter(SERVER, REQUEST_N).count()).isEqualTo(1); assertThat(findCounter(SERVER, REQUEST_RESPONSE).count()).isEqualTo(1); assertThat(findCounter(SERVER, REQUEST_STREAM).count()).isEqualTo(1); assertThat(findCounter(SERVER, SETUP).count()).isEqualTo(1); }