@Test public void updateSubscriptionForFilteredEntities_badFilterType() { final ChannelSchema channelSchema = new ChannelSchema( 0, ValueUtil.randomString(), null, ChannelSchema.FilterType.STATIC, null, true, true, Collections.emptyList() ); final SystemSchema schema = new SystemSchema( 1, ValueUtil.randomString(), new ChannelSchema[]{ channelSchema }, new EntitySchema[ 0 ] ); final Connector connector = createConnector( schema ); newConnection( connector ); final ChannelAddress address1 = new ChannelAddress( 1, 0, 1 ); final Subscription subscription1 = createSubscription( address1, ValueUtil.randomString(), true ); final IllegalStateException exception = expectThrows( IllegalStateException.class, () -> safeAction( () -> connector.updateSubscriptionForFilteredEntities( subscription1 ) ) ); assertEquals( exception.getMessage(), "Replicant-0079: Connector.updateSubscriptionForFilteredEntities invoked for address 1.0.1 but the channel does not have a DYNAMIC filter." ); }
@SuppressWarnings( "unchecked" ) void updateSubscriptionForFilteredEntities( @Nonnull final Subscription subscription ) { final ChannelAddress address = subscription.getAddress(); final ChannelSchema channel = getSchema().getChannel( address.getChannelId() ); if ( Replicant.shouldCheckInvariants() ) { invariant( () -> ChannelSchema.FilterType.DYNAMIC == channel.getFilterType(), () -> "Replicant-0079: Connector.updateSubscriptionForFilteredEntities invoked for address " + subscription.getAddress() + " but the channel does not have a DYNAMIC filter." ); } final List<Entity> entitiesToDelink = new ArrayList<>(); for ( final Class<?> entityType : new ArrayList<>( subscription.findAllEntityTypes() ) ) { final List<Entity> entities = subscription.findAllEntitiesByType( entityType ); if ( !entities.isEmpty() ) { @SuppressWarnings( "rawtypes" ) final SubscriptionUpdateEntityFilter updateFilter = channel.getFilter(); assert null != updateFilter; final Object filter = subscription.getFilter(); for ( final Entity entity : entities ) { if ( !updateFilter.doesEntityMatchFilter( filter, entity ) ) { entitiesToDelink.add( entity ); } } } } for ( final Entity entity : entitiesToDelink ) { entity.delinkFromSubscription( subscription ); } }
Connector extends ReplicantService { @SuppressWarnings( "unchecked" ) void updateSubscriptionForFilteredEntities( @Nonnull final Subscription subscription ) { final ChannelAddress address = subscription.getAddress(); final ChannelSchema channel = getSchema().getChannel( address.getChannelId() ); if ( Replicant.shouldCheckInvariants() ) { invariant( () -> ChannelSchema.FilterType.DYNAMIC == channel.getFilterType(), () -> "Replicant-0079: Connector.updateSubscriptionForFilteredEntities invoked for address " + subscription.getAddress() + " but the channel does not have a DYNAMIC filter." ); } final List<Entity> entitiesToDelink = new ArrayList<>(); for ( final Class<?> entityType : new ArrayList<>( subscription.findAllEntityTypes() ) ) { final List<Entity> entities = subscription.findAllEntitiesByType( entityType ); if ( !entities.isEmpty() ) { @SuppressWarnings( "rawtypes" ) final SubscriptionUpdateEntityFilter updateFilter = channel.getFilter(); assert null != updateFilter; final Object filter = subscription.getFilter(); for ( final Entity entity : entities ) { if ( !updateFilter.doesEntityMatchFilter( filter, entity ) ) { entitiesToDelink.add( entity ); } } } } for ( final Entity entity : entitiesToDelink ) { entity.delinkFromSubscription( subscription ); } } }
Connector extends ReplicantService { @SuppressWarnings( "unchecked" ) void updateSubscriptionForFilteredEntities( @Nonnull final Subscription subscription ) { final ChannelAddress address = subscription.getAddress(); final ChannelSchema channel = getSchema().getChannel( address.getChannelId() ); if ( Replicant.shouldCheckInvariants() ) { invariant( () -> ChannelSchema.FilterType.DYNAMIC == channel.getFilterType(), () -> "Replicant-0079: Connector.updateSubscriptionForFilteredEntities invoked for address " + subscription.getAddress() + " but the channel does not have a DYNAMIC filter." ); } final List<Entity> entitiesToDelink = new ArrayList<>(); for ( final Class<?> entityType : new ArrayList<>( subscription.findAllEntityTypes() ) ) { final List<Entity> entities = subscription.findAllEntitiesByType( entityType ); if ( !entities.isEmpty() ) { @SuppressWarnings( "rawtypes" ) final SubscriptionUpdateEntityFilter updateFilter = channel.getFilter(); assert null != updateFilter; final Object filter = subscription.getFilter(); for ( final Entity entity : entities ) { if ( !updateFilter.doesEntityMatchFilter( filter, entity ) ) { entitiesToDelink.add( entity ); } } } } for ( final Entity entity : entitiesToDelink ) { entity.delinkFromSubscription( subscription ); } } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); }
Connector extends ReplicantService { @SuppressWarnings( "unchecked" ) void updateSubscriptionForFilteredEntities( @Nonnull final Subscription subscription ) { final ChannelAddress address = subscription.getAddress(); final ChannelSchema channel = getSchema().getChannel( address.getChannelId() ); if ( Replicant.shouldCheckInvariants() ) { invariant( () -> ChannelSchema.FilterType.DYNAMIC == channel.getFilterType(), () -> "Replicant-0079: Connector.updateSubscriptionForFilteredEntities invoked for address " + subscription.getAddress() + " but the channel does not have a DYNAMIC filter." ); } final List<Entity> entitiesToDelink = new ArrayList<>(); for ( final Class<?> entityType : new ArrayList<>( subscription.findAllEntityTypes() ) ) { final List<Entity> entities = subscription.findAllEntitiesByType( entityType ); if ( !entities.isEmpty() ) { @SuppressWarnings( "rawtypes" ) final SubscriptionUpdateEntityFilter updateFilter = channel.getFilter(); assert null != updateFilter; final Object filter = subscription.getFilter(); for ( final Entity entity : entities ) { if ( !updateFilter.doesEntityMatchFilter( filter, entity ) ) { entitiesToDelink.add( entity ); } } } } for ( final Entity entity : entitiesToDelink ) { entity.delinkFromSubscription( subscription ); } } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); @Override String toString(); }
Connector extends ReplicantService { @SuppressWarnings( "unchecked" ) void updateSubscriptionForFilteredEntities( @Nonnull final Subscription subscription ) { final ChannelAddress address = subscription.getAddress(); final ChannelSchema channel = getSchema().getChannel( address.getChannelId() ); if ( Replicant.shouldCheckInvariants() ) { invariant( () -> ChannelSchema.FilterType.DYNAMIC == channel.getFilterType(), () -> "Replicant-0079: Connector.updateSubscriptionForFilteredEntities invoked for address " + subscription.getAddress() + " but the channel does not have a DYNAMIC filter." ); } final List<Entity> entitiesToDelink = new ArrayList<>(); for ( final Class<?> entityType : new ArrayList<>( subscription.findAllEntityTypes() ) ) { final List<Entity> entities = subscription.findAllEntitiesByType( entityType ); if ( !entities.isEmpty() ) { @SuppressWarnings( "rawtypes" ) final SubscriptionUpdateEntityFilter updateFilter = channel.getFilter(); assert null != updateFilter; final Object filter = subscription.getFilter(); for ( final Entity entity : entities ) { if ( !updateFilter.doesEntityMatchFilter( filter, entity ) ) { entitiesToDelink.add( entity ); } } } } for ( final Entity entity : entitiesToDelink ) { entity.delinkFromSubscription( subscription ); } } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); @Override String toString();  }