@Test public void linkSubscriptionEntries() throws Exception { final ChannelMetaData ch1 = new ChannelMetaData( 0, "Roster", null, ChannelMetaData.FilterType.DYNAMIC, String.class, ChannelMetaData.CacheType.NONE, false, true ); final ChannelMetaData ch2 = new ChannelMetaData( 1, "Resource", 42, ChannelMetaData.FilterType.NONE, null, ChannelMetaData.CacheType.NONE, false, true ); final ChannelMetaData[] channels = new ChannelMetaData[]{ ch1, ch2 }; final ChannelAddress address1 = new ChannelAddress( ch1.getChannelId(), null ); final ChannelAddress address2a = new ChannelAddress( ch2.getChannelId(), ValueUtil.randomInt() ); final ChannelAddress address2b = new ChannelAddress( ch2.getChannelId(), ValueUtil.randomInt() ); final TestReplicantSessionManager sm = new TestReplicantSessionManager( channels ); final ReplicantSession session = createSession( sm ); final SubscriptionEntry se1 = session.createSubscriptionEntry( address1 ); final SubscriptionEntry se2a = session.createSubscriptionEntry( address2a ); final SubscriptionEntry se2b = session.createSubscriptionEntry( address2b ); assertEntry( se1, false, 0, 0, null ); assertEntry( se2a, false, 0, 0, null ); assertEntry( se2b, false, 0, 0, null ); with( session, () -> sm.linkSubscriptionEntries( se1, se2a ) ); assertEntry( se1, false, 0, 1, null ); assertEntry( se2a, false, 1, 0, null ); assertEntry( se2b, false, 0, 0, null ); assertTrue( se1.getOutwardSubscriptions().contains( address2a ) ); assertTrue( se2a.getInwardSubscriptions().contains( address1 ) ); with( session, () -> sm.linkSubscriptionEntries( se2a, se2b ) ); assertEntry( se1, false, 0, 1, null ); assertEntry( se2a, false, 1, 1, null ); assertEntry( se2b, false, 1, 0, null ); assertTrue( se2a.getOutwardSubscriptions().contains( address2b ) ); assertTrue( se2b.getInwardSubscriptions().contains( address2a ) ); with( session, () -> sm.delinkSubscriptionEntries( se2a, se2b ) ); assertEntry( se1, false, 0, 1, null ); assertEntry( se2a, false, 1, 0, null ); assertEntry( se2b, false, 0, 0, null ); assertFalse( se2a.getOutwardSubscriptions().contains( address2b ) ); assertFalse( se2b.getInwardSubscriptions().contains( address2a ) ); with( session, () -> sm.delinkSubscriptionEntries( se2a, se2b ) ); assertEntry( se1, false, 0, 1, null ); assertEntry( se2a, false, 1, 0, null ); assertEntry( se2b, false, 0, 0, null ); assertFalse( se2a.getOutwardSubscriptions().contains( address2b ) ); assertFalse( se2b.getInwardSubscriptions().contains( address2a ) ); with( session, () -> sm.delinkSubscriptionEntries( se1, se2a ) ); assertEntry( se1, false, 0, 0, null ); assertEntry( se2a, false, 0, 0, null ); assertEntry( se2b, false, 0, 0, null ); assertFalse( se1.getOutwardSubscriptions().contains( address2a ) ); assertFalse( se2a.getInwardSubscriptions().contains( address1 ) ); }
void linkSubscriptionEntries( @Nonnull final SubscriptionEntry sourceEntry, @Nonnull final SubscriptionEntry targetEntry ) { sourceEntry.registerOutwardSubscriptions( targetEntry.getAddress() ); targetEntry.registerInwardSubscriptions( sourceEntry.getAddress() ); }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { void linkSubscriptionEntries( @Nonnull final SubscriptionEntry sourceEntry, @Nonnull final SubscriptionEntry targetEntry ) { sourceEntry.registerOutwardSubscriptions( targetEntry.getAddress() ); targetEntry.registerInwardSubscriptions( sourceEntry.getAddress() ); } }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { void linkSubscriptionEntries( @Nonnull final SubscriptionEntry sourceEntry, @Nonnull final SubscriptionEntry targetEntry ) { sourceEntry.registerOutwardSubscriptions( targetEntry.getAddress() ); targetEntry.registerInwardSubscriptions( sourceEntry.getAddress() ); }  }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { void linkSubscriptionEntries( @Nonnull final SubscriptionEntry sourceEntry, @Nonnull final SubscriptionEntry targetEntry ) { sourceEntry.registerOutwardSubscriptions( targetEntry.getAddress() ); targetEntry.registerInwardSubscriptions( sourceEntry.getAddress() ); }  @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 { void linkSubscriptionEntries( @Nonnull final SubscriptionEntry sourceEntry, @Nonnull final SubscriptionEntry targetEntry ) { sourceEntry.registerOutwardSubscriptions( targetEntry.getAddress() ); targetEntry.registerInwardSubscriptions( sourceEntry.getAddress() ); }  @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 );  }