@Test public void progressAreaOfInterestRequestProcessing_Update() { final ChannelSchema channelSchema = new ChannelSchema( 0, ValueUtil.randomString(), String.class, ChannelSchema.FilterType.DYNAMIC, mock( SubscriptionUpdateEntityFilter.class ), false, true, Collections.emptyList() ); final SystemSchema schema = new SystemSchema( 1, ValueUtil.randomString(), new ChannelSchema[]{ channelSchema }, new EntitySchema[]{} ); final Connector connector = createConnector( schema ); final Connection connection = newConnection( connector ); final ChannelAddress address1 = new ChannelAddress( 1, 0, 1 ); final String filter = ValueUtil.randomString(); final AreaOfInterestRequest request1 = new AreaOfInterestRequest( address1, AreaOfInterestRequest.Type.UPDATE, filter ); pauseScheduler(); createSubscription( address1, null, true ); connection.injectCurrentAreaOfInterestRequest( request1 ); final TestSpyEventHandler handler = registerTestSpyEventHandler(); assertFalse( connection.getCurrentAreaOfInterestRequests().isEmpty() ); connector.progressAreaOfInterestRequestProcessing(); assertFalse( connection.getCurrentAreaOfInterestRequests().isEmpty() ); handler.assertEventCount( 1 ); handler.assertNextEvent( SubscriptionUpdateStartedEvent.class, e -> assertEquals( e.getAddress(), address1 ) ); }
boolean progressAreaOfInterestRequestProcessing() { final List<AreaOfInterestRequest> requests = new ArrayList<>( ensureConnection().getCurrentAreaOfInterestRequests() ); if ( requests.isEmpty() ) { return false; } else if ( requests.get( 0 ).isInProgress() ) { return false; } else { final AreaOfInterestRequest.Type type = requests.get( 0 ).getType(); if ( AreaOfInterestRequest.Type.ADD == type ) { progressAreaOfInterestAddRequests( requests ); } else if ( AreaOfInterestRequest.Type.REMOVE == type ) { progressAreaOfInterestRemoveRequests( requests ); } else { progressAreaOfInterestUpdateRequests( requests ); } return true; } }
Connector extends ReplicantService { boolean progressAreaOfInterestRequestProcessing() { final List<AreaOfInterestRequest> requests = new ArrayList<>( ensureConnection().getCurrentAreaOfInterestRequests() ); if ( requests.isEmpty() ) { return false; } else if ( requests.get( 0 ).isInProgress() ) { return false; } else { final AreaOfInterestRequest.Type type = requests.get( 0 ).getType(); if ( AreaOfInterestRequest.Type.ADD == type ) { progressAreaOfInterestAddRequests( requests ); } else if ( AreaOfInterestRequest.Type.REMOVE == type ) { progressAreaOfInterestRemoveRequests( requests ); } else { progressAreaOfInterestUpdateRequests( requests ); } return true; } } }
Connector extends ReplicantService { boolean progressAreaOfInterestRequestProcessing() { final List<AreaOfInterestRequest> requests = new ArrayList<>( ensureConnection().getCurrentAreaOfInterestRequests() ); if ( requests.isEmpty() ) { return false; } else if ( requests.get( 0 ).isInProgress() ) { return false; } else { final AreaOfInterestRequest.Type type = requests.get( 0 ).getType(); if ( AreaOfInterestRequest.Type.ADD == type ) { progressAreaOfInterestAddRequests( requests ); } else if ( AreaOfInterestRequest.Type.REMOVE == type ) { progressAreaOfInterestRemoveRequests( requests ); } else { progressAreaOfInterestUpdateRequests( requests ); } return true; } } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); }
Connector extends ReplicantService { boolean progressAreaOfInterestRequestProcessing() { final List<AreaOfInterestRequest> requests = new ArrayList<>( ensureConnection().getCurrentAreaOfInterestRequests() ); if ( requests.isEmpty() ) { return false; } else if ( requests.get( 0 ).isInProgress() ) { return false; } else { final AreaOfInterestRequest.Type type = requests.get( 0 ).getType(); if ( AreaOfInterestRequest.Type.ADD == type ) { progressAreaOfInterestAddRequests( requests ); } else if ( AreaOfInterestRequest.Type.REMOVE == type ) { progressAreaOfInterestRemoveRequests( requests ); } else { progressAreaOfInterestUpdateRequests( requests ); } return true; } } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); @Override String toString(); }
Connector extends ReplicantService { boolean progressAreaOfInterestRequestProcessing() { final List<AreaOfInterestRequest> requests = new ArrayList<>( ensureConnection().getCurrentAreaOfInterestRequests() ); if ( requests.isEmpty() ) { return false; } else if ( requests.get( 0 ).isInProgress() ) { return false; } else { final AreaOfInterestRequest.Type type = requests.get( 0 ).getType(); if ( AreaOfInterestRequest.Type.ADD == type ) { progressAreaOfInterestAddRequests( requests ); } else if ( AreaOfInterestRequest.Type.REMOVE == type ) { progressAreaOfInterestRemoveRequests( requests ); } else { progressAreaOfInterestUpdateRequests( requests ); } return true; } } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); @Override String toString();  }