@Test public void expandLinkIfRequired() throws Exception { EntityMessageCacheUtil.removeSessionChanges(); final ChannelMetaData ch1 = new ChannelMetaData( 0, "C1", null, ChannelMetaData.FilterType.DYNAMIC, String.class, ChannelMetaData.CacheType.NONE, false, true ); final ChannelMetaData ch2 = new ChannelMetaData( 1, "C2", 42, ChannelMetaData.FilterType.NONE, null, ChannelMetaData.CacheType.NONE, false, true ); final ChannelMetaData ch3 = new ChannelMetaData( 2, "C3", 42, ChannelMetaData.FilterType.DYNAMIC, String.class, ChannelMetaData.CacheType.NONE, false, true ); final ChannelMetaData[] channels = new ChannelMetaData[]{ ch1, ch2, ch3 }; 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 ChannelAddress address3a = new ChannelAddress( ch3.getChannelId(), ValueUtil.randomInt() ); final ChannelAddress address3b = new ChannelAddress( ch3.getChannelId(), ValueUtil.randomInt() ); final ChannelLink link1 = new ChannelLink( address1, address2a ); final ChannelLink link2 = new ChannelLink( address1, address2b ); final ChannelLink link3 = new ChannelLink( address1, address3a ); final ChannelLink link4 = new ChannelLink( address1, address3b ); final TestReplicantSessionManager sm = new TestReplicantSessionManager( channels ); final ReplicantSession session = createSession( sm ); with( session, () -> assertFalse( sm.expandLinkIfRequired( session, link1, EntityMessageCacheUtil.getSessionChanges() ) ) ); with( session, () -> sm.subscribe( session, address1, new TestFilter( 33 ) ) ); final SubscriptionEntry entry1 = with( session, () -> session.getSubscriptionEntry( address1 ) ); assertEquals( entry1.getInwardSubscriptions().size(), 0 ); assertEquals( entry1.getOutwardSubscriptions().size(), 0 ); with( session, () -> assertTrue( sm.expandLinkIfRequired( session, link1, EntityMessageCacheUtil.getSessionChanges() ) ) ); assertEquals( entry1.getInwardSubscriptions().size(), 0 ); assertEquals( entry1.getOutwardSubscriptions().size(), 1 ); final SubscriptionEntry entry2a = with( session, () -> session.getSubscriptionEntry( address2a ) ); assertFalse( entry2a.isExplicitlySubscribed() ); assertEquals( entry2a.getInwardSubscriptions().size(), 1 ); assertEquals( entry2a.getOutwardSubscriptions().size(), 0 ); with( session, () -> assertFalse( sm.expandLinkIfRequired( session, link1, EntityMessageCacheUtil.getSessionChanges() ) ) ); with( session, () -> sm.subscribe( session, address2b, null ) ); final SubscriptionEntry entry2b = with( session, () -> session.getSubscriptionEntry( address2b ) ); assertEquals( entry1.getInwardSubscriptions().size(), 0 ); assertEquals( entry1.getOutwardSubscriptions().size(), 1 ); assertEquals( entry2b.getInwardSubscriptions().size(), 0 ); assertEquals( entry2b.getOutwardSubscriptions().size(), 0 ); with( session, () -> assertFalse( sm.expandLinkIfRequired( session, link2, EntityMessageCacheUtil.getSessionChanges() ) ) ); assertEquals( entry1.getInwardSubscriptions().size(), 0 ); assertEquals( entry1.getOutwardSubscriptions().size(), 2 ); assertEquals( entry2b.getInwardSubscriptions().size(), 1 ); assertEquals( entry2b.getOutwardSubscriptions().size(), 0 ); with( session, () -> assertFalse( sm.expandLinkIfRequired( session, link3, EntityMessageCacheUtil.getSessionChanges() ) ) ); sm.setFollowSource( link3.getSourceChannel() ); with( session, () -> assertTrue( sm.expandLinkIfRequired( session, link3, EntityMessageCacheUtil.getSessionChanges() ) ) ); final SubscriptionEntry entry3a = with( session, () -> session.getSubscriptionEntry( address3a ) ); assertEquals( entry1.getInwardSubscriptions().size(), 0 ); assertEquals( entry1.getOutwardSubscriptions().size(), 3 ); assertEquals( entry3a.getInwardSubscriptions().size(), 1 ); assertEquals( entry3a.getOutwardSubscriptions().size(), 0 ); assertEquals( entry3a.getFilter(), entry1.getFilter() ); with( session, () -> sm.subscribe( session, address3b, new TestFilter( ValueUtil.randomInt() ) ) ); final SubscriptionEntry entry3b = with( session, () -> session.getSubscriptionEntry( address3b ) ); with( session, () -> assertFalse( sm.expandLinkIfRequired( session, link4, EntityMessageCacheUtil.getSessionChanges() ) ) ); assertEquals( entry1.getInwardSubscriptions().size(), 0 ); assertEquals( entry1.getOutwardSubscriptions().size(), 4 ); assertEquals( entry3b.getInwardSubscriptions().size(), 1 ); assertEquals( entry3b.getOutwardSubscriptions().size(), 0 ); assertNotEquals( entry3b.getFilter(), entry1.getFilter() ); }
boolean expandLinkIfRequired( @Nonnull final ReplicantSession session, @Nonnull final ChannelLink link, @Nonnull final ChangeSet changeSet ) { final ChannelAddress source = link.getSourceChannel(); final SubscriptionEntry sourceEntry = session.findSubscriptionEntry( source ); if ( null != sourceEntry ) { final ChannelAddress target = link.getTargetChannel(); final boolean linkingConditional = !getSystemMetaData().getChannelMetaData( target ).hasFilterParameter(); if ( linkingConditional || shouldFollowLink( sourceEntry, target ) ) { final SubscriptionEntry targetEntry = session.findSubscriptionEntry( target ); if ( null == targetEntry ) { subscribe( session, target, false, linkingConditional ? null : sourceEntry.getFilter(), changeSet ); linkSubscriptionEntries( sourceEntry, session.getSubscriptionEntry( target ) ); return true; } else { linkSubscriptionEntries( sourceEntry, targetEntry ); } } } return false; }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { boolean expandLinkIfRequired( @Nonnull final ReplicantSession session, @Nonnull final ChannelLink link, @Nonnull final ChangeSet changeSet ) { final ChannelAddress source = link.getSourceChannel(); final SubscriptionEntry sourceEntry = session.findSubscriptionEntry( source ); if ( null != sourceEntry ) { final ChannelAddress target = link.getTargetChannel(); final boolean linkingConditional = !getSystemMetaData().getChannelMetaData( target ).hasFilterParameter(); if ( linkingConditional || shouldFollowLink( sourceEntry, target ) ) { final SubscriptionEntry targetEntry = session.findSubscriptionEntry( target ); if ( null == targetEntry ) { subscribe( session, target, false, linkingConditional ? null : sourceEntry.getFilter(), changeSet ); linkSubscriptionEntries( sourceEntry, session.getSubscriptionEntry( target ) ); return true; } else { linkSubscriptionEntries( sourceEntry, targetEntry ); } } } return false; } }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { boolean expandLinkIfRequired( @Nonnull final ReplicantSession session, @Nonnull final ChannelLink link, @Nonnull final ChangeSet changeSet ) { final ChannelAddress source = link.getSourceChannel(); final SubscriptionEntry sourceEntry = session.findSubscriptionEntry( source ); if ( null != sourceEntry ) { final ChannelAddress target = link.getTargetChannel(); final boolean linkingConditional = !getSystemMetaData().getChannelMetaData( target ).hasFilterParameter(); if ( linkingConditional || shouldFollowLink( sourceEntry, target ) ) { final SubscriptionEntry targetEntry = session.findSubscriptionEntry( target ); if ( null == targetEntry ) { subscribe( session, target, false, linkingConditional ? null : sourceEntry.getFilter(), changeSet ); linkSubscriptionEntries( sourceEntry, session.getSubscriptionEntry( target ) ); return true; } else { linkSubscriptionEntries( sourceEntry, targetEntry ); } } } return false; }  }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { boolean expandLinkIfRequired( @Nonnull final ReplicantSession session, @Nonnull final ChannelLink link, @Nonnull final ChangeSet changeSet ) { final ChannelAddress source = link.getSourceChannel(); final SubscriptionEntry sourceEntry = session.findSubscriptionEntry( source ); if ( null != sourceEntry ) { final ChannelAddress target = link.getTargetChannel(); final boolean linkingConditional = !getSystemMetaData().getChannelMetaData( target ).hasFilterParameter(); if ( linkingConditional || shouldFollowLink( sourceEntry, target ) ) { final SubscriptionEntry targetEntry = session.findSubscriptionEntry( target ); if ( null == targetEntry ) { subscribe( session, target, false, linkingConditional ? null : sourceEntry.getFilter(), changeSet ); linkSubscriptionEntries( sourceEntry, session.getSubscriptionEntry( target ) ); return true; } else { linkSubscriptionEntries( sourceEntry, targetEntry ); } } } return false; }  @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 { boolean expandLinkIfRequired( @Nonnull final ReplicantSession session, @Nonnull final ChannelLink link, @Nonnull final ChangeSet changeSet ) { final ChannelAddress source = link.getSourceChannel(); final SubscriptionEntry sourceEntry = session.findSubscriptionEntry( source ); if ( null != sourceEntry ) { final ChannelAddress target = link.getTargetChannel(); final boolean linkingConditional = !getSystemMetaData().getChannelMetaData( target ).hasFilterParameter(); if ( linkingConditional || shouldFollowLink( sourceEntry, target ) ) { final SubscriptionEntry targetEntry = session.findSubscriptionEntry( target ); if ( null == targetEntry ) { subscribe( session, target, false, linkingConditional ? null : sourceEntry.getFilter(), changeSet ); linkSubscriptionEntries( sourceEntry, session.getSubscriptionEntry( target ) ); return true; } else { linkSubscriptionEntries( sourceEntry, targetEntry ); } } } return false; }  @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 );  }