@Test public void performSubscribe() throws Exception { final ChannelMetaData ch1 = new ChannelMetaData( 0, "C1", null, ChannelMetaData.FilterType.NONE, null, ChannelMetaData.CacheType.NONE, false, true ); final ChannelMetaData ch2 = new ChannelMetaData( 1, "C2", null, ChannelMetaData.FilterType.DYNAMIC, String.class, ChannelMetaData.CacheType.NONE, false, true ); final ChannelMetaData ch3 = new ChannelMetaData( 2, "C3", 42, ChannelMetaData.FilterType.NONE, null, ChannelMetaData.CacheType.NONE, false, true ); final ChannelMetaData[] channels = new ChannelMetaData[]{ ch1, ch2, ch3 }; final ChannelAddress address1 = new ChannelAddress( ch1.getChannelId() ); final ChannelAddress address2 = new ChannelAddress( ch2.getChannelId() ); final ChannelAddress address3 = new ChannelAddress( ch2.getChannelId(), ValueUtil.randomInt() ); final TestReplicantSessionManager sm = new TestReplicantSessionManager( channels ); final ReplicantSession session = createSession( sm ); { final SubscriptionEntry e1 = session.createSubscriptionEntry( address1 ); EntityMessageCacheUtil.removeSessionChanges(); assertChannelActionCount( 0 ); assertEntry( e1, false, 0, 0, null ); with( session, () -> sm.performSubscribe( session, e1, true, null, EntityMessageCacheUtil.getSessionChanges() ) ); assertEntry( e1, true, 0, 0, null ); final List<ChannelAction> actions = getChannelActions(); assertEquals( actions.size(), 1 ); assertChannelAction( actions.get( 0 ), address1, ChannelAction.Action.ADD, null ); final Collection<Change> changes = getChanges(); assertEquals( changes.size(), 1 ); assertEquals( changes.iterator().next().getEntityMessage().getId(), 79 ); assertEntry( e1, true, 0, 0, null ); } { final TestFilter filter = new TestFilter( 42 ); EntityMessageCacheUtil.removeSessionChanges(); final SubscriptionEntry e1 = session.createSubscriptionEntry( address2 ); assertChannelActionCount( 0 ); assertEntry( e1, false, 0, 0, null ); with( session, () -> sm.performSubscribe( session, e1, true, filter, EntityMessageCacheUtil.getSessionChanges() ) ); assertEntry( e1, true, 0, 0, filter ); final List<ChannelAction> actions = getChannelActions(); assertEquals( actions.size(), 1 ); assertChannelAction( actions.get( 0 ), address2, ChannelAction.Action.ADD, "{\"myField\":42}" ); final Collection<Change> changes = getChanges(); assertEquals( changes.size(), 1 ); assertEquals( changes.iterator().next().getEntityMessage().getId(), 79 ); assertEntry( e1, true, 0, 0, filter ); with( session, () -> session.deleteSubscriptionEntry( e1 ) ); } { final SubscriptionEntry e1 = session.createSubscriptionEntry( address3 ); EntityMessageCacheUtil.removeSessionChanges(); assertChannelActionCount( 0 ); assertEntry( e1, false, 0, 0, null ); sm.markChannelRootAsDeleted(); with( session, () -> sm.performSubscribe( session, e1, true, null, EntityMessageCacheUtil.getSessionChanges() ) ); assertEntry( e1, false, 0, 0, null ); final List<ChannelAction> actions = getChannelActions(); assertEquals( actions.size(), 1 ); assertChannelAction( actions.get( 0 ), address3, ChannelAction.Action.DELETE, null ); assertEquals( getChanges().size(), 0 ); } }
void performSubscribe( @Nonnull final ReplicantSession session, @Nonnull final SubscriptionEntry entry, final boolean explicitSubscribe, @Nullable final Object filter, @Nonnull final ChangeSet changeSet ) { entry.setFilter( filter ); final ChannelAddress address = entry.getAddress(); final ChannelMetaData channelMetaData = getSystemMetaData().getChannelMetaData( address ); if ( channelMetaData.isCacheable() ) { final ChannelCacheEntry cacheEntry = tryGetCacheEntry( address ); if ( null != cacheEntry ) { if ( explicitSubscribe ) { entry.setExplicitlySubscribed( true ); } final String eTag = cacheEntry.getCacheKey(); if ( eTag.equals( session.getETag( address ) ) ) { if ( session.getWebSocketSession().isOpen() ) { final JsonObjectBuilder response = Json .createObjectBuilder() .add( "type", "use-cache" ) .add( "channel", address.toString() ) .add( "etag", eTag ); final Integer requestId = (Integer) getRegistry().getResource( ServerConstants.REQUEST_ID_KEY ); if ( null != requestId ) { response.add( "requestId", requestId ); } WebSocketUtil.sendJsonObject( session.getWebSocketSession(), response.build() ); changeSet.setRequired( false ); getRegistry().putResource( ServerConstants.CACHED_RESULT_HANDLED_KEY, "1" ); } } else { session.setETag( address, null ); final ChangeSet cacheChangeSet = new ChangeSet(); cacheChangeSet.merge( cacheEntry.getChangeSet(), true ); cacheChangeSet.mergeAction( address, ChannelAction.Action.ADD, filter ); queueCachedChangeSet( session, eTag, cacheChangeSet ); changeSet.setRequired( false ); } return; } else { assert address.hasSubChannelId(); final ChangeSet cacheChangeSet = new ChangeSet(); cacheChangeSet.mergeAction( address, ChannelAction.Action.DELETE, null ); queueCachedChangeSet( session, null, cacheChangeSet ); changeSet.setRequired( false ); return; } } final SubscribeResult result = collectDataForSubscribe( address, changeSet, filter ); if ( result.isChannelRootDeleted() ) { changeSet.mergeAction( address, ChannelAction.Action.DELETE, null ); } else { changeSet.mergeAction( address, ChannelAction.Action.ADD, filter ); if ( explicitSubscribe ) { entry.setExplicitlySubscribed( true ); } } }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { void performSubscribe( @Nonnull final ReplicantSession session, @Nonnull final SubscriptionEntry entry, final boolean explicitSubscribe, @Nullable final Object filter, @Nonnull final ChangeSet changeSet ) { entry.setFilter( filter ); final ChannelAddress address = entry.getAddress(); final ChannelMetaData channelMetaData = getSystemMetaData().getChannelMetaData( address ); if ( channelMetaData.isCacheable() ) { final ChannelCacheEntry cacheEntry = tryGetCacheEntry( address ); if ( null != cacheEntry ) { if ( explicitSubscribe ) { entry.setExplicitlySubscribed( true ); } final String eTag = cacheEntry.getCacheKey(); if ( eTag.equals( session.getETag( address ) ) ) { if ( session.getWebSocketSession().isOpen() ) { final JsonObjectBuilder response = Json .createObjectBuilder() .add( "type", "use-cache" ) .add( "channel", address.toString() ) .add( "etag", eTag ); final Integer requestId = (Integer) getRegistry().getResource( ServerConstants.REQUEST_ID_KEY ); if ( null != requestId ) { response.add( "requestId", requestId ); } WebSocketUtil.sendJsonObject( session.getWebSocketSession(), response.build() ); changeSet.setRequired( false ); getRegistry().putResource( ServerConstants.CACHED_RESULT_HANDLED_KEY, "1" ); } } else { session.setETag( address, null ); final ChangeSet cacheChangeSet = new ChangeSet(); cacheChangeSet.merge( cacheEntry.getChangeSet(), true ); cacheChangeSet.mergeAction( address, ChannelAction.Action.ADD, filter ); queueCachedChangeSet( session, eTag, cacheChangeSet ); changeSet.setRequired( false ); } return; } else { assert address.hasSubChannelId(); final ChangeSet cacheChangeSet = new ChangeSet(); cacheChangeSet.mergeAction( address, ChannelAction.Action.DELETE, null ); queueCachedChangeSet( session, null, cacheChangeSet ); changeSet.setRequired( false ); return; } } final SubscribeResult result = collectDataForSubscribe( address, changeSet, filter ); if ( result.isChannelRootDeleted() ) { changeSet.mergeAction( address, ChannelAction.Action.DELETE, null ); } else { changeSet.mergeAction( address, ChannelAction.Action.ADD, filter ); if ( explicitSubscribe ) { entry.setExplicitlySubscribed( true ); } } } }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { void performSubscribe( @Nonnull final ReplicantSession session, @Nonnull final SubscriptionEntry entry, final boolean explicitSubscribe, @Nullable final Object filter, @Nonnull final ChangeSet changeSet ) { entry.setFilter( filter ); final ChannelAddress address = entry.getAddress(); final ChannelMetaData channelMetaData = getSystemMetaData().getChannelMetaData( address ); if ( channelMetaData.isCacheable() ) { final ChannelCacheEntry cacheEntry = tryGetCacheEntry( address ); if ( null != cacheEntry ) { if ( explicitSubscribe ) { entry.setExplicitlySubscribed( true ); } final String eTag = cacheEntry.getCacheKey(); if ( eTag.equals( session.getETag( address ) ) ) { if ( session.getWebSocketSession().isOpen() ) { final JsonObjectBuilder response = Json .createObjectBuilder() .add( "type", "use-cache" ) .add( "channel", address.toString() ) .add( "etag", eTag ); final Integer requestId = (Integer) getRegistry().getResource( ServerConstants.REQUEST_ID_KEY ); if ( null != requestId ) { response.add( "requestId", requestId ); } WebSocketUtil.sendJsonObject( session.getWebSocketSession(), response.build() ); changeSet.setRequired( false ); getRegistry().putResource( ServerConstants.CACHED_RESULT_HANDLED_KEY, "1" ); } } else { session.setETag( address, null ); final ChangeSet cacheChangeSet = new ChangeSet(); cacheChangeSet.merge( cacheEntry.getChangeSet(), true ); cacheChangeSet.mergeAction( address, ChannelAction.Action.ADD, filter ); queueCachedChangeSet( session, eTag, cacheChangeSet ); changeSet.setRequired( false ); } return; } else { assert address.hasSubChannelId(); final ChangeSet cacheChangeSet = new ChangeSet(); cacheChangeSet.mergeAction( address, ChannelAction.Action.DELETE, null ); queueCachedChangeSet( session, null, cacheChangeSet ); changeSet.setRequired( false ); return; } } final SubscribeResult result = collectDataForSubscribe( address, changeSet, filter ); if ( result.isChannelRootDeleted() ) { changeSet.mergeAction( address, ChannelAction.Action.DELETE, null ); } else { changeSet.mergeAction( address, ChannelAction.Action.ADD, filter ); if ( explicitSubscribe ) { entry.setExplicitlySubscribed( true ); } } }  }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { void performSubscribe( @Nonnull final ReplicantSession session, @Nonnull final SubscriptionEntry entry, final boolean explicitSubscribe, @Nullable final Object filter, @Nonnull final ChangeSet changeSet ) { entry.setFilter( filter ); final ChannelAddress address = entry.getAddress(); final ChannelMetaData channelMetaData = getSystemMetaData().getChannelMetaData( address ); if ( channelMetaData.isCacheable() ) { final ChannelCacheEntry cacheEntry = tryGetCacheEntry( address ); if ( null != cacheEntry ) { if ( explicitSubscribe ) { entry.setExplicitlySubscribed( true ); } final String eTag = cacheEntry.getCacheKey(); if ( eTag.equals( session.getETag( address ) ) ) { if ( session.getWebSocketSession().isOpen() ) { final JsonObjectBuilder response = Json .createObjectBuilder() .add( "type", "use-cache" ) .add( "channel", address.toString() ) .add( "etag", eTag ); final Integer requestId = (Integer) getRegistry().getResource( ServerConstants.REQUEST_ID_KEY ); if ( null != requestId ) { response.add( "requestId", requestId ); } WebSocketUtil.sendJsonObject( session.getWebSocketSession(), response.build() ); changeSet.setRequired( false ); getRegistry().putResource( ServerConstants.CACHED_RESULT_HANDLED_KEY, "1" ); } } else { session.setETag( address, null ); final ChangeSet cacheChangeSet = new ChangeSet(); cacheChangeSet.merge( cacheEntry.getChangeSet(), true ); cacheChangeSet.mergeAction( address, ChannelAction.Action.ADD, filter ); queueCachedChangeSet( session, eTag, cacheChangeSet ); changeSet.setRequired( false ); } return; } else { assert address.hasSubChannelId(); final ChangeSet cacheChangeSet = new ChangeSet(); cacheChangeSet.mergeAction( address, ChannelAction.Action.DELETE, null ); queueCachedChangeSet( session, null, cacheChangeSet ); changeSet.setRequired( false ); return; } } final SubscribeResult result = collectDataForSubscribe( address, changeSet, filter ); if ( result.isChannelRootDeleted() ) { changeSet.mergeAction( address, ChannelAction.Action.DELETE, null ); } else { changeSet.mergeAction( address, ChannelAction.Action.ADD, filter ); if ( explicitSubscribe ) { entry.setExplicitlySubscribed( true ); } } }  @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 performSubscribe( @Nonnull final ReplicantSession session, @Nonnull final SubscriptionEntry entry, final boolean explicitSubscribe, @Nullable final Object filter, @Nonnull final ChangeSet changeSet ) { entry.setFilter( filter ); final ChannelAddress address = entry.getAddress(); final ChannelMetaData channelMetaData = getSystemMetaData().getChannelMetaData( address ); if ( channelMetaData.isCacheable() ) { final ChannelCacheEntry cacheEntry = tryGetCacheEntry( address ); if ( null != cacheEntry ) { if ( explicitSubscribe ) { entry.setExplicitlySubscribed( true ); } final String eTag = cacheEntry.getCacheKey(); if ( eTag.equals( session.getETag( address ) ) ) { if ( session.getWebSocketSession().isOpen() ) { final JsonObjectBuilder response = Json .createObjectBuilder() .add( "type", "use-cache" ) .add( "channel", address.toString() ) .add( "etag", eTag ); final Integer requestId = (Integer) getRegistry().getResource( ServerConstants.REQUEST_ID_KEY ); if ( null != requestId ) { response.add( "requestId", requestId ); } WebSocketUtil.sendJsonObject( session.getWebSocketSession(), response.build() ); changeSet.setRequired( false ); getRegistry().putResource( ServerConstants.CACHED_RESULT_HANDLED_KEY, "1" ); } } else { session.setETag( address, null ); final ChangeSet cacheChangeSet = new ChangeSet(); cacheChangeSet.merge( cacheEntry.getChangeSet(), true ); cacheChangeSet.mergeAction( address, ChannelAction.Action.ADD, filter ); queueCachedChangeSet( session, eTag, cacheChangeSet ); changeSet.setRequired( false ); } return; } else { assert address.hasSubChannelId(); final ChangeSet cacheChangeSet = new ChangeSet(); cacheChangeSet.mergeAction( address, ChannelAction.Action.DELETE, null ); queueCachedChangeSet( session, null, cacheChangeSet ); changeSet.setRequired( false ); return; } } final SubscribeResult result = collectDataForSubscribe( address, changeSet, filter ); if ( result.isChannelRootDeleted() ) { changeSet.mergeAction( address, ChannelAction.Action.DELETE, null ); } else { changeSet.mergeAction( address, ChannelAction.Action.ADD, filter ); if ( explicitSubscribe ) { entry.setExplicitlySubscribed( true ); } } }  @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 );  }