@Test public void requestSubscriptionUpdate() { final ChannelSchema channelSchema = new ChannelSchema( 0, ValueUtil.randomString(), null, ChannelSchema.FilterType.DYNAMIC, ( f, e ) -> true, false, true, Collections.emptyList() ); final SystemSchema schema = new SystemSchema( 1, ValueUtil.randomString(), new ChannelSchema[]{ channelSchema }, new EntitySchema[ 0 ] ); final Connector connector = createConnector( schema ); newConnection( connector ); connector.pauseMessageScheduler(); final ChannelAddress address = new ChannelAddress( 1, 0 ); assertFalse( connector.isAreaOfInterestRequestPending( AreaOfInterestRequest.Type.UPDATE, address, null ) ); final TestSpyEventHandler handler = registerTestSpyEventHandler(); connector.requestSubscriptionUpdate( address, null ); assertTrue( connector.isAreaOfInterestRequestPending( AreaOfInterestRequest.Type.UPDATE, address, null ) ); handler.assertEventCount( 1 ); handler.assertNextEvent( SubscriptionUpdateRequestQueuedEvent.class, e -> assertEquals( e.getAddress(), address ) ); }
void requestSubscriptionUpdate( @Nonnull final ChannelAddress address, @Nullable final Object filter ) { if ( Replicant.shouldCheckInvariants() ) { invariant( () -> ChannelSchema.FilterType.DYNAMIC == getSchema().getChannel( address.getChannelId() ).getFilterType(), () -> "Replicant-0082: Connector.requestSubscriptionUpdate invoked for channel " + address + " but channel does not have a dynamic filter." ); } ensureConnection().requestSubscriptionUpdate( address, filter ); triggerMessageScheduler(); if ( Replicant.areSpiesEnabled() && getReplicantContext().getSpy().willPropagateSpyEvents() ) { getReplicantContext().getSpy().reportSpyEvent( new SubscriptionUpdateRequestQueuedEvent( address, filter ) ); } }
Connector extends ReplicantService { void requestSubscriptionUpdate( @Nonnull final ChannelAddress address, @Nullable final Object filter ) { if ( Replicant.shouldCheckInvariants() ) { invariant( () -> ChannelSchema.FilterType.DYNAMIC == getSchema().getChannel( address.getChannelId() ).getFilterType(), () -> "Replicant-0082: Connector.requestSubscriptionUpdate invoked for channel " + address + " but channel does not have a dynamic filter." ); } ensureConnection().requestSubscriptionUpdate( address, filter ); triggerMessageScheduler(); if ( Replicant.areSpiesEnabled() && getReplicantContext().getSpy().willPropagateSpyEvents() ) { getReplicantContext().getSpy().reportSpyEvent( new SubscriptionUpdateRequestQueuedEvent( address, filter ) ); } } }
Connector extends ReplicantService { void requestSubscriptionUpdate( @Nonnull final ChannelAddress address, @Nullable final Object filter ) { if ( Replicant.shouldCheckInvariants() ) { invariant( () -> ChannelSchema.FilterType.DYNAMIC == getSchema().getChannel( address.getChannelId() ).getFilterType(), () -> "Replicant-0082: Connector.requestSubscriptionUpdate invoked for channel " + address + " but channel does not have a dynamic filter." ); } ensureConnection().requestSubscriptionUpdate( address, filter ); triggerMessageScheduler(); if ( Replicant.areSpiesEnabled() && getReplicantContext().getSpy().willPropagateSpyEvents() ) { getReplicantContext().getSpy().reportSpyEvent( new SubscriptionUpdateRequestQueuedEvent( address, filter ) ); } } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); }
Connector extends ReplicantService { void requestSubscriptionUpdate( @Nonnull final ChannelAddress address, @Nullable final Object filter ) { if ( Replicant.shouldCheckInvariants() ) { invariant( () -> ChannelSchema.FilterType.DYNAMIC == getSchema().getChannel( address.getChannelId() ).getFilterType(), () -> "Replicant-0082: Connector.requestSubscriptionUpdate invoked for channel " + address + " but channel does not have a dynamic filter." ); } ensureConnection().requestSubscriptionUpdate( address, filter ); triggerMessageScheduler(); if ( Replicant.areSpiesEnabled() && getReplicantContext().getSpy().willPropagateSpyEvents() ) { getReplicantContext().getSpy().reportSpyEvent( new SubscriptionUpdateRequestQueuedEvent( address, filter ) ); } } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); @Override String toString(); }
Connector extends ReplicantService { void requestSubscriptionUpdate( @Nonnull final ChannelAddress address, @Nullable final Object filter ) { if ( Replicant.shouldCheckInvariants() ) { invariant( () -> ChannelSchema.FilterType.DYNAMIC == getSchema().getChannel( address.getChannelId() ).getFilterType(), () -> "Replicant-0082: Connector.requestSubscriptionUpdate invoked for channel " + address + " but channel does not have a dynamic filter." ); } ensureConnection().requestSubscriptionUpdate( address, filter ); triggerMessageScheduler(); if ( Replicant.areSpiesEnabled() && getReplicantContext().getSpy().willPropagateSpyEvents() ) { getReplicantContext().getSpy().reportSpyEvent( new SubscriptionUpdateRequestQueuedEvent( address, filter ) ); } } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); @Override String toString();  }