SignalFilter { @SuppressWarnings("squid:S3864") List<Target> filter(final Signal<?> signal) { return connection.getTargets().stream() .filter(t -> isTargetAuthorized(t, signal)) .peek(authorizedTarget -> connectionMonitorRegistry.forOutboundDispatched(connection.getId(), authorizedTarget.getAddress()) .success(signal)) .filter(t -> isTargetSubscribedForTopic(t, signal)) .peek(filteredTarget -> connectionMonitorRegistry.forOutboundFiltered(connection.getId(), filteredTarget.getAddress()) .success(signal)) .collect(Collectors.toList()); } SignalFilter(final Connection connection,
            final ConnectionMonitorRegistry<ConnectionMonitor> connectionMonitorRegistry);   }
@Test public void test() { final Connection connection = ConnectivityModelFactory.newConnectionBuilder(CONNECTION_ID, ConnectionType.AMQP_10, ConnectivityStatus.OPEN, URI).targets(targets).build(); final SignalFilter signalFilter = new SignalFilter(connection, connectionMonitorRegistry); final List<Target> filteredTargets = signalFilter.filter(signal(signalTopic, readSubjects)); Assertions.assertThat(filteredTargets) .isEqualTo(expectedTargets); }