@Test public void removeOrphanSubscription() { final Connector connector = createConnector(); newConnection( connector ); pauseScheduler(); connector.pauseMessageScheduler(); final ChannelAddress address = new ChannelAddress( 1, 0 ); safeAction( () -> { final Subscription subscription = createSubscription( address, null, true ); connector.setState( ConnectorState.CONNECTED ); final TestSpyEventHandler handler = registerTestSpyEventHandler(); Replicant.context().getConverger().removeOrphanSubscriptions(); final List<AreaOfInterestRequest> requests = connector.ensureConnection().getPendingAreaOfInterestRequests(); assertEquals( requests.size(), 1 ); final AreaOfInterestRequest request = requests.get( 0 ); assertEquals( request.getType(), AreaOfInterestRequest.Type.REMOVE ); assertEquals( request.getAddress(), address ); handler.assertEventCount( 2 ); handler.assertNextEvent( SubscriptionOrphanedEvent.class, e -> assertEquals( e.getSubscription(), subscription ) ); handler.assertNextEvent( UnsubscribeRequestQueuedEvent.class, e -> assertEquals( e.getAddress(), address ) ); } ); }
private void removeOrphanSubscription( @Nonnull final ChannelAddress address ) { if ( Replicant.areSpiesEnabled() && getReplicantContext().getSpy().willPropagateSpyEvents() ) { final Subscription subscription = getReplicantContext().findSubscription( address ); assert null != subscription; getReplicantContext().getSpy().reportSpyEvent( new SubscriptionOrphanedEvent( subscription ) ); } getReplicantRuntime().getConnector( address.getSystemId() ).requestUnsubscribe( address ); }
Converger extends ReplicantService { private void removeOrphanSubscription( @Nonnull final ChannelAddress address ) { if ( Replicant.areSpiesEnabled() && getReplicantContext().getSpy().willPropagateSpyEvents() ) { final Subscription subscription = getReplicantContext().findSubscription( address ); assert null != subscription; getReplicantContext().getSpy().reportSpyEvent( new SubscriptionOrphanedEvent( subscription ) ); } getReplicantRuntime().getConnector( address.getSystemId() ).requestUnsubscribe( address ); } }
Converger extends ReplicantService { private void removeOrphanSubscription( @Nonnull final ChannelAddress address ) { if ( Replicant.areSpiesEnabled() && getReplicantContext().getSpy().willPropagateSpyEvents() ) { final Subscription subscription = getReplicantContext().findSubscription( address ); assert null != subscription; getReplicantContext().getSpy().reportSpyEvent( new SubscriptionOrphanedEvent( subscription ) ); } getReplicantRuntime().getConnector( address.getSystemId() ).requestUnsubscribe( address ); } Converger( @Nullable final ReplicantContext context ); }
Converger extends ReplicantService { private void removeOrphanSubscription( @Nonnull final ChannelAddress address ) { if ( Replicant.areSpiesEnabled() && getReplicantContext().getSpy().willPropagateSpyEvents() ) { final Subscription subscription = getReplicantContext().findSubscription( address ); assert null != subscription; getReplicantContext().getSpy().reportSpyEvent( new SubscriptionOrphanedEvent( subscription ) ); } getReplicantRuntime().getConnector( address.getSystemId() ).requestUnsubscribe( address ); } Converger( @Nullable final ReplicantContext context );  }
Converger extends ReplicantService { private void removeOrphanSubscription( @Nonnull final ChannelAddress address ) { if ( Replicant.areSpiesEnabled() && getReplicantContext().getSpy().willPropagateSpyEvents() ) { final Subscription subscription = getReplicantContext().findSubscription( address ); assert null != subscription; getReplicantContext().getSpy().reportSpyEvent( new SubscriptionOrphanedEvent( subscription ) ); } getReplicantRuntime().getConnector( address.getSystemId() ).requestUnsubscribe( address ); } Converger( @Nullable final ReplicantContext context );   }