@Test public void tryGetCacheEntry() { final ChannelMetaData ch1 = new ChannelMetaData( 0, "C1", null, ChannelMetaData.FilterType.NONE, null, ChannelMetaData.CacheType.INTERNAL, false, true ); final ChannelMetaData[] channels = new ChannelMetaData[]{ ch1 }; final ChannelAddress address1 = new ChannelAddress( ch1.getChannelId(), null ); final TestReplicantSessionManager sm = new TestReplicantSessionManager( channels ); final String cacheKey = ValueUtil.randomString(); sm.setCacheKey( cacheKey ); assertFalse( sm.getCacheEntry( address1 ).isInitialized() ); final ChannelCacheEntry entry = sm.tryGetCacheEntry( address1 ); assertNotNull( entry ); assertTrue( entry.isInitialized() ); assertEquals( entry.getDescriptor(), address1 ); assertEquals( entry.getCacheKey(), cacheKey ); assertEquals( entry.getChangeSet().getChanges().size(), 1 ); }
@Nullable ChannelCacheEntry tryGetCacheEntry( @Nonnull final ChannelAddress address ) { assert getSystemMetaData().getChannelMetaData( address ).isCacheable(); final ChannelCacheEntry entry = getCacheEntry( address ); entry.getLock().readLock().lock(); try { if ( entry.isInitialized() ) { return entry; } } finally { entry.getLock().readLock().unlock(); } entry.getLock().writeLock().lock(); try { if ( entry.isInitialized() ) { return entry; } final ChangeSet changeSet = new ChangeSet(); final SubscribeResult result = collectDataForSubscribe( address, changeSet, null ); if ( result.isChannelRootDeleted() ) { return null; } else { final String cacheKey = result.getCacheKey(); assert null != cacheKey; entry.init( cacheKey, changeSet ); return entry; } } finally { entry.getLock().writeLock().unlock(); } }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { @Nullable ChannelCacheEntry tryGetCacheEntry( @Nonnull final ChannelAddress address ) { assert getSystemMetaData().getChannelMetaData( address ).isCacheable(); final ChannelCacheEntry entry = getCacheEntry( address ); entry.getLock().readLock().lock(); try { if ( entry.isInitialized() ) { return entry; } } finally { entry.getLock().readLock().unlock(); } entry.getLock().writeLock().lock(); try { if ( entry.isInitialized() ) { return entry; } final ChangeSet changeSet = new ChangeSet(); final SubscribeResult result = collectDataForSubscribe( address, changeSet, null ); if ( result.isChannelRootDeleted() ) { return null; } else { final String cacheKey = result.getCacheKey(); assert null != cacheKey; entry.init( cacheKey, changeSet ); return entry; } } finally { entry.getLock().writeLock().unlock(); } } }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { @Nullable ChannelCacheEntry tryGetCacheEntry( @Nonnull final ChannelAddress address ) { assert getSystemMetaData().getChannelMetaData( address ).isCacheable(); final ChannelCacheEntry entry = getCacheEntry( address ); entry.getLock().readLock().lock(); try { if ( entry.isInitialized() ) { return entry; } } finally { entry.getLock().readLock().unlock(); } entry.getLock().writeLock().lock(); try { if ( entry.isInitialized() ) { return entry; } final ChangeSet changeSet = new ChangeSet(); final SubscribeResult result = collectDataForSubscribe( address, changeSet, null ); if ( result.isChannelRootDeleted() ) { return null; } else { final String cacheKey = result.getCacheKey(); assert null != cacheKey; entry.init( cacheKey, changeSet ); return entry; } } finally { entry.getLock().writeLock().unlock(); } }  }
ReplicantSessionManagerImpl implements EntityMessageEndpoint, ReplicantSessionManager { @Nullable ChannelCacheEntry tryGetCacheEntry( @Nonnull final ChannelAddress address ) { assert getSystemMetaData().getChannelMetaData( address ).isCacheable(); final ChannelCacheEntry entry = getCacheEntry( address ); entry.getLock().readLock().lock(); try { if ( entry.isInitialized() ) { return entry; } } finally { entry.getLock().readLock().unlock(); } entry.getLock().writeLock().lock(); try { if ( entry.isInitialized() ) { return entry; } final ChangeSet changeSet = new ChangeSet(); final SubscribeResult result = collectDataForSubscribe( address, changeSet, null ); if ( result.isChannelRootDeleted() ) { return null; } else { final String cacheKey = result.getCacheKey(); assert null != cacheKey; entry.init( cacheKey, changeSet ); return entry; } } finally { entry.getLock().writeLock().unlock(); } }  @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 { @Nullable ChannelCacheEntry tryGetCacheEntry( @Nonnull final ChannelAddress address ) { assert getSystemMetaData().getChannelMetaData( address ).isCacheable(); final ChannelCacheEntry entry = getCacheEntry( address ); entry.getLock().readLock().lock(); try { if ( entry.isInitialized() ) { return entry; } } finally { entry.getLock().readLock().unlock(); } entry.getLock().writeLock().lock(); try { if ( entry.isInitialized() ) { return entry; } final ChangeSet changeSet = new ChangeSet(); final SubscribeResult result = collectDataForSubscribe( address, changeSet, null ); if ( result.isChannelRootDeleted() ) { return null; } else { final String cacheKey = result.getCacheKey(); assert null != cacheKey; entry.init( cacheKey, changeSet ); return entry; } } finally { entry.getLock().writeLock().unlock(); } }  @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 );  }