@Test public void onMessageReceived() { final Connector connector = createConnector(); final Connection connection = newConnection( connector ); pauseScheduler(); connector.pauseMessageScheduler(); assertEquals( connection.getPendingResponses().size(), 0 ); assertFalse( connector.isSchedulerActive() ); final ChangeSetMessage message = ChangeSetMessage.create( null, null, null, null, null ); connector.onMessageReceived( message ); assertEquals( connection.getPendingResponses().size(), 1 ); assertEquals( connection.getPendingResponses().get( 0 ).getMessage(), message ); assertTrue( connector.isSchedulerActive() ); }
void onMessageReceived( @Nonnull final ServerToClientMessage message ) { final Connection connection = ensureConnection(); final Integer requestId = message.getRequestId(); final RequestEntry request = null != requestId ? connection.getRequest( requestId ) : null; @Nonnull final ServerToClientMessage messageToQueue; if ( UseCacheMessage.TYPE.equals( message.getType() ) ) { final UseCacheMessage useCacheMessage = (UseCacheMessage) message; final String channel = useCacheMessage.getChannel(); final ChannelAddress address = ChannelAddress.parse( getSchema().getId(), channel ); final String etag = useCacheMessage.getEtag(); final CacheService cacheService = getReplicantContext().getCacheService(); if ( Replicant.shouldCheckApiInvariants() ) { apiInvariant( () -> null != cacheService, () -> "Replicant-0042: Received a use-cache message for channel " + address + " but no cache service configured." ); } assert null != cacheService; final CacheEntry entry = cacheService.lookup( address ); if ( Replicant.shouldCheckApiInvariants() ) { apiInvariant( () -> null != entry, () -> "Replicant-0068: Received a use-cache message for channel " + channel + " but no cache entry present for channel." ); assert null != entry; apiInvariant( () -> entry.getETag().equals( etag ), () -> "Replicant-0075: Received a use-cache message for channel " + channel + " with etag '" + etag + "' but cache entry has etag '" + entry.getETag() + "'." ); } assert null != entry; messageToQueue = Js.cast( Global.JSON.parse( entry.getContent() ) ); messageToQueue.setRequestId( requestId ); } else { messageToQueue = message; } connection.enqueueResponse( messageToQueue, request ); triggerMessageScheduler(); }
Connector extends ReplicantService { void onMessageReceived( @Nonnull final ServerToClientMessage message ) { final Connection connection = ensureConnection(); final Integer requestId = message.getRequestId(); final RequestEntry request = null != requestId ? connection.getRequest( requestId ) : null; @Nonnull final ServerToClientMessage messageToQueue; if ( UseCacheMessage.TYPE.equals( message.getType() ) ) { final UseCacheMessage useCacheMessage = (UseCacheMessage) message; final String channel = useCacheMessage.getChannel(); final ChannelAddress address = ChannelAddress.parse( getSchema().getId(), channel ); final String etag = useCacheMessage.getEtag(); final CacheService cacheService = getReplicantContext().getCacheService(); if ( Replicant.shouldCheckApiInvariants() ) { apiInvariant( () -> null != cacheService, () -> "Replicant-0042: Received a use-cache message for channel " + address + " but no cache service configured." ); } assert null != cacheService; final CacheEntry entry = cacheService.lookup( address ); if ( Replicant.shouldCheckApiInvariants() ) { apiInvariant( () -> null != entry, () -> "Replicant-0068: Received a use-cache message for channel " + channel + " but no cache entry present for channel." ); assert null != entry; apiInvariant( () -> entry.getETag().equals( etag ), () -> "Replicant-0075: Received a use-cache message for channel " + channel + " with etag '" + etag + "' but cache entry has etag '" + entry.getETag() + "'." ); } assert null != entry; messageToQueue = Js.cast( Global.JSON.parse( entry.getContent() ) ); messageToQueue.setRequestId( requestId ); } else { messageToQueue = message; } connection.enqueueResponse( messageToQueue, request ); triggerMessageScheduler(); } }
Connector extends ReplicantService { void onMessageReceived( @Nonnull final ServerToClientMessage message ) { final Connection connection = ensureConnection(); final Integer requestId = message.getRequestId(); final RequestEntry request = null != requestId ? connection.getRequest( requestId ) : null; @Nonnull final ServerToClientMessage messageToQueue; if ( UseCacheMessage.TYPE.equals( message.getType() ) ) { final UseCacheMessage useCacheMessage = (UseCacheMessage) message; final String channel = useCacheMessage.getChannel(); final ChannelAddress address = ChannelAddress.parse( getSchema().getId(), channel ); final String etag = useCacheMessage.getEtag(); final CacheService cacheService = getReplicantContext().getCacheService(); if ( Replicant.shouldCheckApiInvariants() ) { apiInvariant( () -> null != cacheService, () -> "Replicant-0042: Received a use-cache message for channel " + address + " but no cache service configured." ); } assert null != cacheService; final CacheEntry entry = cacheService.lookup( address ); if ( Replicant.shouldCheckApiInvariants() ) { apiInvariant( () -> null != entry, () -> "Replicant-0068: Received a use-cache message for channel " + channel + " but no cache entry present for channel." ); assert null != entry; apiInvariant( () -> entry.getETag().equals( etag ), () -> "Replicant-0075: Received a use-cache message for channel " + channel + " with etag '" + etag + "' but cache entry has etag '" + entry.getETag() + "'." ); } assert null != entry; messageToQueue = Js.cast( Global.JSON.parse( entry.getContent() ) ); messageToQueue.setRequestId( requestId ); } else { messageToQueue = message; } connection.enqueueResponse( messageToQueue, request ); triggerMessageScheduler(); } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); }
Connector extends ReplicantService { void onMessageReceived( @Nonnull final ServerToClientMessage message ) { final Connection connection = ensureConnection(); final Integer requestId = message.getRequestId(); final RequestEntry request = null != requestId ? connection.getRequest( requestId ) : null; @Nonnull final ServerToClientMessage messageToQueue; if ( UseCacheMessage.TYPE.equals( message.getType() ) ) { final UseCacheMessage useCacheMessage = (UseCacheMessage) message; final String channel = useCacheMessage.getChannel(); final ChannelAddress address = ChannelAddress.parse( getSchema().getId(), channel ); final String etag = useCacheMessage.getEtag(); final CacheService cacheService = getReplicantContext().getCacheService(); if ( Replicant.shouldCheckApiInvariants() ) { apiInvariant( () -> null != cacheService, () -> "Replicant-0042: Received a use-cache message for channel " + address + " but no cache service configured." ); } assert null != cacheService; final CacheEntry entry = cacheService.lookup( address ); if ( Replicant.shouldCheckApiInvariants() ) { apiInvariant( () -> null != entry, () -> "Replicant-0068: Received a use-cache message for channel " + channel + " but no cache entry present for channel." ); assert null != entry; apiInvariant( () -> entry.getETag().equals( etag ), () -> "Replicant-0075: Received a use-cache message for channel " + channel + " with etag '" + etag + "' but cache entry has etag '" + entry.getETag() + "'." ); } assert null != entry; messageToQueue = Js.cast( Global.JSON.parse( entry.getContent() ) ); messageToQueue.setRequestId( requestId ); } else { messageToQueue = message; } connection.enqueueResponse( messageToQueue, request ); triggerMessageScheduler(); } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); @Override String toString(); }
Connector extends ReplicantService { void onMessageReceived( @Nonnull final ServerToClientMessage message ) { final Connection connection = ensureConnection(); final Integer requestId = message.getRequestId(); final RequestEntry request = null != requestId ? connection.getRequest( requestId ) : null; @Nonnull final ServerToClientMessage messageToQueue; if ( UseCacheMessage.TYPE.equals( message.getType() ) ) { final UseCacheMessage useCacheMessage = (UseCacheMessage) message; final String channel = useCacheMessage.getChannel(); final ChannelAddress address = ChannelAddress.parse( getSchema().getId(), channel ); final String etag = useCacheMessage.getEtag(); final CacheService cacheService = getReplicantContext().getCacheService(); if ( Replicant.shouldCheckApiInvariants() ) { apiInvariant( () -> null != cacheService, () -> "Replicant-0042: Received a use-cache message for channel " + address + " but no cache service configured." ); } assert null != cacheService; final CacheEntry entry = cacheService.lookup( address ); if ( Replicant.shouldCheckApiInvariants() ) { apiInvariant( () -> null != entry, () -> "Replicant-0068: Received a use-cache message for channel " + channel + " but no cache entry present for channel." ); assert null != entry; apiInvariant( () -> entry.getETag().equals( etag ), () -> "Replicant-0075: Received a use-cache message for channel " + channel + " with etag '" + etag + "' but cache entry has etag '" + entry.getETag() + "'." ); } assert null != entry; messageToQueue = Js.cast( Global.JSON.parse( entry.getContent() ) ); messageToQueue.setRequestId( requestId ); } else { messageToQueue = message; } connection.enqueueResponse( messageToQueue, request ); triggerMessageScheduler(); } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); @Override String toString();  }