@Test public void updateSubscription() throws Exception { final ChannelMetaData ch = new ChannelMetaData( 0, "C2", null, ChannelMetaData.FilterType.DYNAMIC, String.class, ChannelMetaData.CacheType.NONE, false, true ); final ChannelMetaData[] channels = new ChannelMetaData[]{ ch }; final ChannelAddress cd = new ChannelAddress( ch.getChannelId(), null ); final TestReplicantSessionManager sm = new TestReplicantSessionManager( channels ); final ReplicantSession session = createSession( sm ); final TestFilter originalFilter = new TestFilter( 41 ); final TestFilter filter = new TestFilter( 42 ); EntityMessageCacheUtil.removeSessionChanges(); with( session, () -> sm.subscribe( session, cd, originalFilter ) ); EntityMessageCacheUtil.removeSessionChanges(); TransactionSynchronizationRegistryUtil.lookup() .putResource( ServerConstants.REQUEST_ID_KEY, ValueUtil.randomInt() ); with( session, () -> sm.subscribe( session, cd, originalFilter ) ); final SubscriptionEntry e1 = with( session, () -> session.getSubscriptionEntry( cd ) ); assertEntry( e1, true, 0, 0, originalFilter ); assertChannelActionCount( 0 ); assertSessionChangesCount( 0 ); with( session, () -> sm.subscribe( session, cd, filter ) ); assertEntry( e1, true, 0, 0, filter ); assertChannelActionCount( 1 ); assertSessionChangesCount( 1 ); }
private void updateSubscription( @Nonnull final ReplicantSession session, @Nonnull final ChannelAddress address, @Nullable final Object filter, @Nonnull final ChangeSet changeSet ) { final ChannelMetaData channel = getSystemMetaData().getChannelMetaData( address ); assert channel.hasFilterParameter(); assert channel.getFilterType() == ChannelMetaData.FilterType.DYNAMIC; final SubscriptionEntry entry = session.getSubscriptionEntry( address ); final Object originalFilter = entry.getFilter(); if ( doFiltersNotMatch( filter, originalFilter ) ) { entry.setFilter( filter ); collectDataForSubscriptionUpdate( session, address, changeSet, originalFilter, filter ); changeSet.mergeAction( address, ChannelAction.Action.UPDATE, filter ); for ( final ChannelAction channelAction : new ArrayList<>( changeSet.getChannelActions() ) ) { if ( ChannelAction.Action.REMOVE == channelAction.getAction() ) { final SubscriptionEntry other = session.findSubscriptionEntry( channelAction.getAddress() ); if ( null != other ) { performUnsubscribe( session, other, true, false, changeSet ); } } } propagateSubscriptionFilterUpdate( session, address, filter, changeSet ); } }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { private void updateSubscription( @Nonnull final ReplicantSession session, @Nonnull final ChannelAddress address, @Nullable final Object filter, @Nonnull final ChangeSet changeSet ) { final ChannelMetaData channel = getSystemMetaData().getChannelMetaData( address ); assert channel.hasFilterParameter(); assert channel.getFilterType() == ChannelMetaData.FilterType.DYNAMIC; final SubscriptionEntry entry = session.getSubscriptionEntry( address ); final Object originalFilter = entry.getFilter(); if ( doFiltersNotMatch( filter, originalFilter ) ) { entry.setFilter( filter ); collectDataForSubscriptionUpdate( session, address, changeSet, originalFilter, filter ); changeSet.mergeAction( address, ChannelAction.Action.UPDATE, filter ); for ( final ChannelAction channelAction : new ArrayList<>( changeSet.getChannelActions() ) ) { if ( ChannelAction.Action.REMOVE == channelAction.getAction() ) { final SubscriptionEntry other = session.findSubscriptionEntry( channelAction.getAddress() ); if ( null != other ) { performUnsubscribe( session, other, true, false, changeSet ); } } } propagateSubscriptionFilterUpdate( session, address, filter, changeSet ); } } }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { private void updateSubscription( @Nonnull final ReplicantSession session, @Nonnull final ChannelAddress address, @Nullable final Object filter, @Nonnull final ChangeSet changeSet ) { final ChannelMetaData channel = getSystemMetaData().getChannelMetaData( address ); assert channel.hasFilterParameter(); assert channel.getFilterType() == ChannelMetaData.FilterType.DYNAMIC; final SubscriptionEntry entry = session.getSubscriptionEntry( address ); final Object originalFilter = entry.getFilter(); if ( doFiltersNotMatch( filter, originalFilter ) ) { entry.setFilter( filter ); collectDataForSubscriptionUpdate( session, address, changeSet, originalFilter, filter ); changeSet.mergeAction( address, ChannelAction.Action.UPDATE, filter ); for ( final ChannelAction channelAction : new ArrayList<>( changeSet.getChannelActions() ) ) { if ( ChannelAction.Action.REMOVE == channelAction.getAction() ) { final SubscriptionEntry other = session.findSubscriptionEntry( channelAction.getAddress() ); if ( null != other ) { performUnsubscribe( session, other, true, false, changeSet ); } } } propagateSubscriptionFilterUpdate( session, address, filter, changeSet ); } }  }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { private void updateSubscription( @Nonnull final ReplicantSession session, @Nonnull final ChannelAddress address, @Nullable final Object filter, @Nonnull final ChangeSet changeSet ) { final ChannelMetaData channel = getSystemMetaData().getChannelMetaData( address ); assert channel.hasFilterParameter(); assert channel.getFilterType() == ChannelMetaData.FilterType.DYNAMIC; final SubscriptionEntry entry = session.getSubscriptionEntry( address ); final Object originalFilter = entry.getFilter(); if ( doFiltersNotMatch( filter, originalFilter ) ) { entry.setFilter( filter ); collectDataForSubscriptionUpdate( session, address, changeSet, originalFilter, filter ); changeSet.mergeAction( address, ChannelAction.Action.UPDATE, filter ); for ( final ChannelAction channelAction : new ArrayList<>( changeSet.getChannelActions() ) ) { if ( ChannelAction.Action.REMOVE == channelAction.getAction() ) { final SubscriptionEntry other = session.findSubscriptionEntry( channelAction.getAddress() ); if ( null != other ) { performUnsubscribe( session, other, true, false, changeSet ); } } } propagateSubscriptionFilterUpdate( session, address, filter, changeSet ); } }  @Override boolean invalidateSession( @Nonnull final ReplicantSession session ); @Override @Nullable ReplicantSession getSession( @Nonnull final String sessionId ); @Nonnull @Override Set<String> getSessionIDs(); @Override @Nonnull ReplicantSession createSession( @Nonnull final Session webSocketSession ); @SuppressWarnings( { "WeakerAccess", "unused" } ) void pingSessions(); @SuppressWarnings( "WeakerAccess" ) void removeAllSessions(); @SuppressWarnings( "WeakerAccess" ) void removeClosedSessions(); @Override boolean saveEntityMessages( @Nullable final String sessionId,
                                     @Nullable final Integer requestId,
                                     @Nonnull final Collection<EntityMessage> messages,
                                     @Nullable final ChangeSet sessionChanges ); @Override void sendChangeMessage( @Nonnull final ReplicantSession session,
                                 @Nullable final Integer requestId,
                                 @Nullable final String etag,
                                 @Nonnull final Collection<EntityMessage> messages,
                                 @Nonnull final ChangeSet changeSet ); @Override void bulkSubscribe( @Nonnull final ReplicantSession session,
                             final int channelId,
                             @Nonnull final Collection<Integer> subChannelIds,
                             @Nullable final Object filter ); @Override void subscribe( @Nonnull final ReplicantSession session,
                         @Nonnull final ChannelAddress address,
                         @Nullable final Object filter ); @Override void unsubscribe( @Nonnull final ReplicantSession session, @Nonnull final ChannelAddress address ); @Override void bulkUnsubscribe( @Nonnull final ReplicantSession session,
                               final int channelId,
                               @Nonnull final Collection<Integer> subChannelIds ); }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { private void updateSubscription( @Nonnull final ReplicantSession session, @Nonnull final ChannelAddress address, @Nullable final Object filter, @Nonnull final ChangeSet changeSet ) { final ChannelMetaData channel = getSystemMetaData().getChannelMetaData( address ); assert channel.hasFilterParameter(); assert channel.getFilterType() == ChannelMetaData.FilterType.DYNAMIC; final SubscriptionEntry entry = session.getSubscriptionEntry( address ); final Object originalFilter = entry.getFilter(); if ( doFiltersNotMatch( filter, originalFilter ) ) { entry.setFilter( filter ); collectDataForSubscriptionUpdate( session, address, changeSet, originalFilter, filter ); changeSet.mergeAction( address, ChannelAction.Action.UPDATE, filter ); for ( final ChannelAction channelAction : new ArrayList<>( changeSet.getChannelActions() ) ) { if ( ChannelAction.Action.REMOVE == channelAction.getAction() ) { final SubscriptionEntry other = session.findSubscriptionEntry( channelAction.getAddress() ); if ( null != other ) { performUnsubscribe( session, other, true, false, changeSet ); } } } propagateSubscriptionFilterUpdate( session, address, filter, changeSet ); } }  @Override boolean invalidateSession( @Nonnull final ReplicantSession session ); @Override @Nullable ReplicantSession getSession( @Nonnull final String sessionId ); @Nonnull @Override Set<String> getSessionIDs(); @Override @Nonnull ReplicantSession createSession( @Nonnull final Session webSocketSession ); @SuppressWarnings( { "WeakerAccess", "unused" } ) void pingSessions(); @SuppressWarnings( "WeakerAccess" ) void removeAllSessions(); @SuppressWarnings( "WeakerAccess" ) void removeClosedSessions(); @Override boolean saveEntityMessages( @Nullable final String sessionId,
                                     @Nullable final Integer requestId,
                                     @Nonnull final Collection<EntityMessage> messages,
                                     @Nullable final ChangeSet sessionChanges ); @Override void sendChangeMessage( @Nonnull final ReplicantSession session,
                                 @Nullable final Integer requestId,
                                 @Nullable final String etag,
                                 @Nonnull final Collection<EntityMessage> messages,
                                 @Nonnull final ChangeSet changeSet ); @Override void bulkSubscribe( @Nonnull final ReplicantSession session,
                             final int channelId,
                             @Nonnull final Collection<Integer> subChannelIds,
                             @Nullable final Object filter ); @Override void subscribe( @Nonnull final ReplicantSession session,
                         @Nonnull final ChannelAddress address,
                         @Nullable final Object filter ); @Override void unsubscribe( @Nonnull final ReplicantSession session, @Nonnull final ChannelAddress address ); @Override void bulkUnsubscribe( @Nonnull final ReplicantSession session,
                               final int channelId,
                               @Nonnull final Collection<Integer> subChannelIds );  }