@Test public void progressMessages() { final Connector connector = createConnector(); final Connection connection = newConnection( connector ); final String[] channels = { "+0" }; final MessageResponse response = setCurrentMessageResponse( connection, ChangeSetMessage.create( null, null, channels, null, null ) ); response.setParsedChannelChanges( Collections.singletonList( ChannelChangeDescriptor.from( 1, channels[ 0 ] ) ) ); assertNull( connector.getSchedulerLock() ); connector.resumeMessageScheduler(); final boolean result0 = connector.progressMessages(); assertTrue( result0 ); final Disposable schedulerLock0 = connector.getSchedulerLock(); assertNotNull( schedulerLock0 ); final boolean result1 = connector.progressMessages(); assertTrue( result1 ); assertNull( connector.getSchedulerLock() ); assertTrue( Disposable.isDisposed( schedulerLock0 ) ); final boolean result2 = connector.progressMessages(); assertTrue( result2 ); assertNull( connection.getCurrentMessageResponse() ); final boolean result3 = connector.progressMessages(); assertFalse( result3 ); assertNull( connector.getSchedulerLock() ); }
boolean progressMessages() { if ( _schedulerPaused ) { return false; } if ( null == _schedulerLock ) { _schedulerLock = context().pauseScheduler(); if ( Replicant.isChangeBrokerEnabled() ) { _brokerLock = getReplicantContext().getChangeBroker().pause(); } } try { if ( null != _connection && ConnectorState.DISCONNECTING != _state ) { final boolean step1 = progressAreaOfInterestRequestProcessing(); final boolean step2 = progressResponseProcessing(); _schedulerActive = step1 || step2; } else { _schedulerActive = false; callPostMessageResponseActionIfPresent(); } } catch ( final Throwable e ) { onMessageProcessFailure( e ); _schedulerActive = false; releaseSchedulerLock(); return false; } finally { if ( !_schedulerActive ) { releaseSchedulerLock(); } } return _schedulerActive; }
Connector extends ReplicantService { boolean progressMessages() { if ( _schedulerPaused ) { return false; } if ( null == _schedulerLock ) { _schedulerLock = context().pauseScheduler(); if ( Replicant.isChangeBrokerEnabled() ) { _brokerLock = getReplicantContext().getChangeBroker().pause(); } } try { if ( null != _connection && ConnectorState.DISCONNECTING != _state ) { final boolean step1 = progressAreaOfInterestRequestProcessing(); final boolean step2 = progressResponseProcessing(); _schedulerActive = step1 || step2; } else { _schedulerActive = false; callPostMessageResponseActionIfPresent(); } } catch ( final Throwable e ) { onMessageProcessFailure( e ); _schedulerActive = false; releaseSchedulerLock(); return false; } finally { if ( !_schedulerActive ) { releaseSchedulerLock(); } } return _schedulerActive; } }
Connector extends ReplicantService { boolean progressMessages() { if ( _schedulerPaused ) { return false; } if ( null == _schedulerLock ) { _schedulerLock = context().pauseScheduler(); if ( Replicant.isChangeBrokerEnabled() ) { _brokerLock = getReplicantContext().getChangeBroker().pause(); } } try { if ( null != _connection && ConnectorState.DISCONNECTING != _state ) { final boolean step1 = progressAreaOfInterestRequestProcessing(); final boolean step2 = progressResponseProcessing(); _schedulerActive = step1 || step2; } else { _schedulerActive = false; callPostMessageResponseActionIfPresent(); } } catch ( final Throwable e ) { onMessageProcessFailure( e ); _schedulerActive = false; releaseSchedulerLock(); return false; } finally { if ( !_schedulerActive ) { releaseSchedulerLock(); } } return _schedulerActive; } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); }
Connector extends ReplicantService { boolean progressMessages() { if ( _schedulerPaused ) { return false; } if ( null == _schedulerLock ) { _schedulerLock = context().pauseScheduler(); if ( Replicant.isChangeBrokerEnabled() ) { _brokerLock = getReplicantContext().getChangeBroker().pause(); } } try { if ( null != _connection && ConnectorState.DISCONNECTING != _state ) { final boolean step1 = progressAreaOfInterestRequestProcessing(); final boolean step2 = progressResponseProcessing(); _schedulerActive = step1 || step2; } else { _schedulerActive = false; callPostMessageResponseActionIfPresent(); } } catch ( final Throwable e ) { onMessageProcessFailure( e ); _schedulerActive = false; releaseSchedulerLock(); return false; } finally { if ( !_schedulerActive ) { releaseSchedulerLock(); } } return _schedulerActive; } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); @Override String toString(); }
Connector extends ReplicantService { boolean progressMessages() { if ( _schedulerPaused ) { return false; } if ( null == _schedulerLock ) { _schedulerLock = context().pauseScheduler(); if ( Replicant.isChangeBrokerEnabled() ) { _brokerLock = getReplicantContext().getChangeBroker().pause(); } } try { if ( null != _connection && ConnectorState.DISCONNECTING != _state ) { final boolean step1 = progressAreaOfInterestRequestProcessing(); final boolean step2 = progressResponseProcessing(); _schedulerActive = step1 || step2; } else { _schedulerActive = false; callPostMessageResponseActionIfPresent(); } } catch ( final Throwable e ) { onMessageProcessFailure( e ); _schedulerActive = false; releaseSchedulerLock(); return false; } finally { if ( !_schedulerActive ) { releaseSchedulerLock(); } } return _schedulerActive; } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); @Override String toString();  }