@Test public void requestUnsubscribe() throws Exception { final Connector connector = createConnector(); pauseScheduler(); connector.pauseMessageScheduler(); newConnection( connector ); final ChannelAddress address = new ChannelAddress( 1, 0 ); createSubscription( address, null, true ); assertFalse( connector.isAreaOfInterestRequestPending( AreaOfInterestRequest.Type.REMOVE, address, null ) ); final TestSpyEventHandler handler = registerTestSpyEventHandler(); connector.requestUnsubscribe( address ); Thread.sleep( 100 ); assertTrue( connector.isAreaOfInterestRequestPending( AreaOfInterestRequest.Type.REMOVE, address, null ) ); handler.assertEventCount( 1 ); handler.assertNextEvent( UnsubscribeRequestQueuedEvent.class, e -> assertEquals( e.getAddress(), address ) ); }
void requestUnsubscribe( @Nonnull final ChannelAddress address ) { ensureConnection().requestUnsubscribe( address ); triggerMessageScheduler(); if ( Replicant.areSpiesEnabled() && getReplicantContext().getSpy().willPropagateSpyEvents() ) { getReplicantContext().getSpy().reportSpyEvent( new UnsubscribeRequestQueuedEvent( address ) ); } }
Connector extends ReplicantService { void requestUnsubscribe( @Nonnull final ChannelAddress address ) { ensureConnection().requestUnsubscribe( address ); triggerMessageScheduler(); if ( Replicant.areSpiesEnabled() && getReplicantContext().getSpy().willPropagateSpyEvents() ) { getReplicantContext().getSpy().reportSpyEvent( new UnsubscribeRequestQueuedEvent( address ) ); } } }
Connector extends ReplicantService { void requestUnsubscribe( @Nonnull final ChannelAddress address ) { ensureConnection().requestUnsubscribe( address ); triggerMessageScheduler(); if ( Replicant.areSpiesEnabled() && getReplicantContext().getSpy().willPropagateSpyEvents() ) { getReplicantContext().getSpy().reportSpyEvent( new UnsubscribeRequestQueuedEvent( address ) ); } } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); }
Connector extends ReplicantService { void requestUnsubscribe( @Nonnull final ChannelAddress address ) { ensureConnection().requestUnsubscribe( address ); triggerMessageScheduler(); if ( Replicant.areSpiesEnabled() && getReplicantContext().getSpy().willPropagateSpyEvents() ) { getReplicantContext().getSpy().reportSpyEvent( new UnsubscribeRequestQueuedEvent( address ) ); } } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); @Override String toString(); }
Connector extends ReplicantService { void requestUnsubscribe( @Nonnull final ChannelAddress address ) { ensureConnection().requestUnsubscribe( address ); triggerMessageScheduler(); if ( Replicant.areSpiesEnabled() && getReplicantContext().getSpy().willPropagateSpyEvents() ) { getReplicantContext().getSpy().reportSpyEvent( new UnsubscribeRequestQueuedEvent( address ) ); } } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); @Override String toString();  }