@Test public void updateSubscriptionForFilteredEntities() { final SubscriptionUpdateEntityFilter<?> filter = ( f, entity ) -> entity.getId() > 0; final ChannelSchema channelSchema = new ChannelSchema( 0, ValueUtil.randomString(), null, ChannelSchema.FilterType.DYNAMIC, filter, 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 ChannelAddress address2 = new ChannelAddress( 1, 0, 2 ); final Subscription subscription1 = createSubscription( address1, ValueUtil.randomString(), true ); final Subscription subscription2 = createSubscription( address2, ValueUtil.randomString(), true ); final Entity entity1 = findOrCreateEntity( Integer.class, -1 ); final Entity entity2 = findOrCreateEntity( Integer.class, -2 ); final Entity entity3 = findOrCreateEntity( Integer.class, -3 ); final Entity entity4 = findOrCreateEntity( Integer.class, -4 ); final Entity entity5 = findOrCreateEntity( String.class, 5 ); final Entity entity6 = findOrCreateEntity( String.class, 6 ); safeAction( () -> { entity1.linkToSubscription( subscription1 ); entity2.linkToSubscription( subscription1 ); entity3.linkToSubscription( subscription1 ); entity4.linkToSubscription( subscription1 ); entity5.linkToSubscription( subscription1 ); entity6.linkToSubscription( subscription1 ); entity3.linkToSubscription( subscription2 ); entity4.linkToSubscription( subscription2 ); assertEquals( subscription1.getEntities().size(), 2 ); assertEquals( subscription1.findAllEntitiesByType( Integer.class ).size(), 4 ); assertEquals( subscription1.findAllEntitiesByType( String.class ).size(), 2 ); assertEquals( subscription2.getEntities().size(), 1 ); assertEquals( subscription2.findAllEntitiesByType( Integer.class ).size(), 2 ); } ); safeAction( () -> connector.updateSubscriptionForFilteredEntities( subscription1 ) ); safeAction( () -> { assertTrue( Disposable.isDisposed( entity1 ) ); assertTrue( Disposable.isDisposed( entity2 ) ); assertFalse( Disposable.isDisposed( entity3 ) ); assertFalse( Disposable.isDisposed( entity4 ) ); assertFalse( Disposable.isDisposed( entity5 ) ); assertFalse( Disposable.isDisposed( entity6 ) ); assertEquals( subscription1.getEntities().size(), 1 ); assertEquals( subscription1.findAllEntitiesByType( Integer.class ).size(), 0 ); assertEquals( subscription1.findAllEntitiesByType( String.class ).size(), 2 ); assertEquals( subscription2.getEntities().size(), 1 ); assertEquals( subscription2.findAllEntitiesByType( Integer.class ).size(), 2 ); } ); }
@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();  }