@Test public void performUnsubscribe() throws Exception { final ChannelMetaData ch1 = new ChannelMetaData( 0, "C1", 42, ChannelMetaData.FilterType.NONE, null, ChannelMetaData.CacheType.NONE, false, true ); final ChannelMetaData[] channels = new ChannelMetaData[]{ ch1 }; final ChannelAddress address1 = new ChannelAddress( ch1.getChannelId(), ValueUtil.randomInt() ); final ChannelAddress address2 = new ChannelAddress( ch1.getChannelId(), ValueUtil.randomInt() ); final ChannelAddress address3 = new ChannelAddress( ch1.getChannelId(), ValueUtil.randomInt() ); final ChannelAddress address4 = new ChannelAddress( ch1.getChannelId(), ValueUtil.randomInt() ); final ChannelAddress address5 = new ChannelAddress( ch1.getChannelId(), ValueUtil.randomInt() ); final TestReplicantSessionManager sm = new TestReplicantSessionManager( channels ); final ReplicantSession session = createSession( sm ); { EntityMessageCacheUtil.removeSessionChanges(); with( session, () -> sm.subscribe( session, address1, null ) ); final SubscriptionEntry entry = with( session, () -> session.getSubscriptionEntry( address1 ) ); EntityMessageCacheUtil.removeSessionChanges(); assertChannelActionCount( 0 ); with( session, () -> sm.performUnsubscribe( session, entry, true, false, EntityMessageCacheUtil.getSessionChanges() ) ); assertChannelActionCount( 1 ); assertChannelAction( getChannelActions().get( 0 ), address1, ChannelAction.Action.REMOVE, null ); with( session, () -> assertNull( session.findSubscriptionEntry( address1 ) ) ); } { EntityMessageCacheUtil.removeSessionChanges(); with( session, () -> sm.subscribe( session, address1, null ) ); final SubscriptionEntry entry = with( session, () -> session.getSubscriptionEntry( address1 ) ); EntityMessageCacheUtil.removeSessionChanges(); assertChannelActionCount( 0 ); with( session, () -> sm.performUnsubscribe( session, entry, false, false, EntityMessageCacheUtil.getSessionChanges() ) ); assertChannelActionCount( 0 ); with( session, () -> assertNotNull( session.findSubscriptionEntry( address1 ) ) ); with( session, () -> sm.performUnsubscribe( session, entry, true, false, EntityMessageCacheUtil.getSessionChanges() ) ); assertChannelActionCount( 1 ); with( session, () -> assertNull( session.findSubscriptionEntry( address1 ) ) ); } { EntityMessageCacheUtil.removeSessionChanges(); with( session, () -> sm.subscribe( session, address1, false, null, EntityMessageCacheUtil.getSessionChanges() ) ); final SubscriptionEntry entry = with( session, () -> session.getSubscriptionEntry( address1 ) ); EntityMessageCacheUtil.removeSessionChanges(); assertChannelActionCount( 0 ); with( session, () -> sm.performUnsubscribe( session, entry, false, false, EntityMessageCacheUtil.getSessionChanges() ) ); assertChannelActionCount( 1 ); assertChannelAction( getChannelActions().get( 0 ), address1, ChannelAction.Action.REMOVE, null ); with( session, () -> assertNull( session.findSubscriptionEntry( address1 ) ) ); } { EntityMessageCacheUtil.removeSessionChanges(); with( session, () -> sm.subscribe( session, address1, false, null, EntityMessageCacheUtil.getSessionChanges() ) ); final SubscriptionEntry entry = with( session, () -> session.getSubscriptionEntry( address1 ) ); with( session, () -> sm.subscribe( session, address2, false, null, EntityMessageCacheUtil.getSessionChanges() ) ); final SubscriptionEntry entry2 = with( session, () -> session.getSubscriptionEntry( address2 ) ); with( session, () -> sm.linkSubscriptionEntries( entry2, entry ) ); EntityMessageCacheUtil.removeSessionChanges(); assertChannelActionCount( 0 ); with( session, () -> sm.performUnsubscribe( session, entry, false, false, EntityMessageCacheUtil.getSessionChanges() ) ); assertChannelActionCount( 0 ); with( session, () -> assertNotNull( session.findSubscriptionEntry( address1 ) ) ); with( session, () -> sm.delinkSubscriptionEntries( entry2, entry ) ); with( session, () -> sm.performUnsubscribe( session, entry, false, false, EntityMessageCacheUtil.getSessionChanges() ) ); assertChannelActionCount( 1 ); assertChannelAction( getChannelActions().get( 0 ), address1, ChannelAction.Action.REMOVE, null ); with( session, () -> assertNull( session.findSubscriptionEntry( address1 ) ) ); } { EntityMessageCacheUtil.removeSessionChanges(); with( session, () -> sm.subscribe( session, address1, null ) ); with( session, () -> sm.subscribe( session, address2, false, null, EntityMessageCacheUtil.getSessionChanges() ) ); with( session, () -> sm.subscribe( session, address3, false, null, EntityMessageCacheUtil.getSessionChanges() ) ); with( session, () -> sm.subscribe( session, address4, false, null, EntityMessageCacheUtil.getSessionChanges() ) ); with( session, () -> sm.subscribe( session, address5, false, null, EntityMessageCacheUtil.getSessionChanges() ) ); final SubscriptionEntry entry = with( session, () -> session.getSubscriptionEntry( address1 ) ); final SubscriptionEntry entry2 = with( session, () -> session.getSubscriptionEntry( address2 ) ); final SubscriptionEntry entry3 = with( session, () -> session.getSubscriptionEntry( address3 ) ); final SubscriptionEntry entry4 = with( session, () -> session.getSubscriptionEntry( address4 ) ); final SubscriptionEntry entry5 = with( session, () -> session.getSubscriptionEntry( address5 ) ); with( session, () -> sm.linkSubscriptionEntries( entry, entry2 ) ); with( session, () -> sm.linkSubscriptionEntries( entry, entry3 ) ); with( session, () -> sm.linkSubscriptionEntries( entry3, entry4 ) ); with( session, () -> sm.linkSubscriptionEntries( entry4, entry5 ) ); EntityMessageCacheUtil.removeSessionChanges(); assertChannelActionCount( 0 ); with( session, () -> sm.performUnsubscribe( session, entry, true, false, EntityMessageCacheUtil.getSessionChanges() ) ); assertChannelActionCount( 5 ); for ( final ChannelAction action : getChannelActions() ) { assertEquals( action.getAction(), ChannelAction.Action.REMOVE ); } with( session, () -> assertNull( session.findSubscriptionEntry( address1 ) ) ); with( session, () -> assertNull( session.findSubscriptionEntry( address2 ) ) ); with( session, () -> assertNull( session.findSubscriptionEntry( address3 ) ) ); with( session, () -> assertNull( session.findSubscriptionEntry( address4 ) ) ); with( session, () -> assertNull( session.findSubscriptionEntry( address5 ) ) ); } }
private void performUnsubscribe( @Nonnull final ReplicantSession session, @Nonnull final ChannelAddress address, @Nonnull final ChangeSet changeSet ) { final SubscriptionEntry entry = session.findSubscriptionEntry( address ); if ( null != entry ) { performUnsubscribe( session, entry, true, false, changeSet ); } }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { private void performUnsubscribe( @Nonnull final ReplicantSession session, @Nonnull final ChannelAddress address, @Nonnull final ChangeSet changeSet ) { final SubscriptionEntry entry = session.findSubscriptionEntry( address ); if ( null != entry ) { performUnsubscribe( session, entry, true, false, changeSet ); } } }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { private void performUnsubscribe( @Nonnull final ReplicantSession session, @Nonnull final ChannelAddress address, @Nonnull final ChangeSet changeSet ) { final SubscriptionEntry entry = session.findSubscriptionEntry( address ); if ( null != entry ) { performUnsubscribe( session, entry, true, false, changeSet ); } }  }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { private void performUnsubscribe( @Nonnull final ReplicantSession session, @Nonnull final ChannelAddress address, @Nonnull final ChangeSet changeSet ) { final SubscriptionEntry entry = session.findSubscriptionEntry( address ); if ( null != entry ) { performUnsubscribe( session, entry, true, false, 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 performUnsubscribe( @Nonnull final ReplicantSession session, @Nonnull final ChannelAddress address, @Nonnull final ChangeSet changeSet ) { final SubscriptionEntry entry = session.findSubscriptionEntry( address ); if ( null != entry ) { performUnsubscribe( session, entry, true, false, 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 );  }