MicrometerDuplexConnection implements DuplexConnection { @Override public Mono<Void> onClose() { return delegate.onClose().doAfterTerminate(close::increment); } 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("onClose gathers metrics") @Test void onClose() { when(delegate.onClose()).thenReturn(Mono.empty()); new MicrometerDuplexConnection( CLIENT, delegate, meterRegistry, Tag.of("test-key", "test-value")) .onClose() .subscribe(Operators.drainSubscriber()); assertThat( meterRegistry .get("rsocket.duplex.connection.close") .tag("connection.type", CLIENT.name()) .tag("test-key", "test-value") .counter() .count()) .isEqualTo(1); }