MicrometerDuplexConnection implements DuplexConnection { @Override public Flux<ByteBuf> receive() { return delegate.receive().doOnNext(frameCounters); } 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("receive gathers metrics") @Test void receive() { Flux<ByteBuf> frames = Flux.just( createTestCancelFrame(), createTestErrorFrame(), createTestKeepaliveFrame(), createTestLeaseFrame(), createTestMetadataPushFrame(), createTestPayloadFrame(), createTestRequestChannelFrame(), createTestRequestFireAndForgetFrame(), createTestRequestNFrame(), createTestRequestResponseFrame(), createTestRequestStreamFrame(), createTestSetupFrame()); when(delegate.receive()).thenReturn(frames); new MicrometerDuplexConnection( CLIENT, delegate, meterRegistry, Tag.of("test-key", "test-value")) .receive() .as(StepVerifier::create) .expectNextCount(12) .verifyComplete(); assertThat(findCounter(CLIENT, CANCEL).count()).isEqualTo(1); assertThat(findCounter(CLIENT, COMPLETE).count()).isEqualTo(1); assertThat(findCounter(CLIENT, ERROR).count()).isEqualTo(1); assertThat(findCounter(CLIENT, KEEPALIVE).count()).isEqualTo(1); assertThat(findCounter(CLIENT, LEASE).count()).isEqualTo(1); assertThat(findCounter(CLIENT, METADATA_PUSH).count()).isEqualTo(1); assertThat(findCounter(CLIENT, REQUEST_CHANNEL).count()).isEqualTo(1); assertThat(findCounter(CLIENT, REQUEST_FNF).count()).isEqualTo(1); assertThat(findCounter(CLIENT, REQUEST_N).count()).isEqualTo(1); assertThat(findCounter(CLIENT, REQUEST_RESPONSE).count()).isEqualTo(1); assertThat(findCounter(CLIENT, REQUEST_STREAM).count()).isEqualTo(1); assertThat(findCounter(CLIENT, SETUP).count()).isEqualTo(1); }