@Test public void shouldResumeWhileRunning () throws ExecutionException, InterruptedException { thread.start(); runner.resume().get(); }
CompletableFuture<Boolean> resume () { final CompletableFuture<Boolean> result; synchronized ( monitor ) { PipelineReplicaRunnerCommand command = this.command; final PipelineReplicaRunnerStatus status = this.status; if ( command != null ) { final PipelineReplicaRunnerCommandType type = command.type; if ( type == RESUME ) { checkState( status == PAUSED, "%s: cannot be resumed since its status is %s and pending command is %s", id, status, type ); LOGGER.debug( "{}: {} command is already set", id, RESUME ); monitor.notify(); result = command.future; } else if ( type == PAUSE ) { checkState( status == RUNNING, "Pipeline %s cannot be resumed since its status is %s and pending command is %s", id, status, type ); LOGGER.debug( "{}: Completing pending {} command because of new {} command.", id, PAUSE, RESUME ); command.complete(); this.command = null; result = CompletableFuture.completedFuture( TRUE ); } else if ( type == REFRESH ) { LOGGER.debug( "{}: switching pending {} command to {}", id, REFRESH, RESUME ); command = new PipelineReplicaRunnerCommand( RESUME, command.future ); this.command = command; result = command.future; } else { LOGGER.error( "{}: {} failed since there is a pending {} command", id, RESUME, type ); result = new CompletableFuture<>(); command.future.thenRun( () -> result.completeExceptionally( new IllegalStateException( id + ": " + RESUME + " failed since there " + "is a pending " + type + " command" ) ) ); } } else if ( status == RUNNING ) { LOGGER.debug( "{} is already {}", id, RUNNING ); result = CompletableFuture.completedFuture( TRUE ); } else if ( status == PAUSED ) { LOGGER.debug( "{}: {} command is set", id, RESUME ); command = new PipelineReplicaRunnerCommand( RESUME ); this.command = command; result = command.future; } else { LOGGER.error( "{}: {} failed since status is {}", id, RESUME, status ); result = new CompletableFuture<>(); result.completeExceptionally( new IllegalStateException( id + ": " + RESUME + " failed since status is " + status ) ); } } return result; }
PipelineReplicaRunner implements Runnable { CompletableFuture<Boolean> resume () { final CompletableFuture<Boolean> result; synchronized ( monitor ) { PipelineReplicaRunnerCommand command = this.command; final PipelineReplicaRunnerStatus status = this.status; if ( command != null ) { final PipelineReplicaRunnerCommandType type = command.type; if ( type == RESUME ) { checkState( status == PAUSED, "%s: cannot be resumed since its status is %s and pending command is %s", id, status, type ); LOGGER.debug( "{}: {} command is already set", id, RESUME ); monitor.notify(); result = command.future; } else if ( type == PAUSE ) { checkState( status == RUNNING, "Pipeline %s cannot be resumed since its status is %s and pending command is %s", id, status, type ); LOGGER.debug( "{}: Completing pending {} command because of new {} command.", id, PAUSE, RESUME ); command.complete(); this.command = null; result = CompletableFuture.completedFuture( TRUE ); } else if ( type == REFRESH ) { LOGGER.debug( "{}: switching pending {} command to {}", id, REFRESH, RESUME ); command = new PipelineReplicaRunnerCommand( RESUME, command.future ); this.command = command; result = command.future; } else { LOGGER.error( "{}: {} failed since there is a pending {} command", id, RESUME, type ); result = new CompletableFuture<>(); command.future.thenRun( () -> result.completeExceptionally( new IllegalStateException( id + ": " + RESUME + " failed since there " + "is a pending " + type + " command" ) ) ); } } else if ( status == RUNNING ) { LOGGER.debug( "{} is already {}", id, RUNNING ); result = CompletableFuture.completedFuture( TRUE ); } else if ( status == PAUSED ) { LOGGER.debug( "{}: {} command is set", id, RESUME ); command = new PipelineReplicaRunnerCommand( RESUME ); this.command = command; result = command.future; } else { LOGGER.error( "{}: {} failed since status is {}", id, RESUME, status ); result = new CompletableFuture<>(); result.completeExceptionally( new IllegalStateException( id + ": " + RESUME + " failed since status is " + status ) ); } } return result; } }
PipelineReplicaRunner implements Runnable { CompletableFuture<Boolean> resume () { final CompletableFuture<Boolean> result; synchronized ( monitor ) { PipelineReplicaRunnerCommand command = this.command; final PipelineReplicaRunnerStatus status = this.status; if ( command != null ) { final PipelineReplicaRunnerCommandType type = command.type; if ( type == RESUME ) { checkState( status == PAUSED, "%s: cannot be resumed since its status is %s and pending command is %s", id, status, type ); LOGGER.debug( "{}: {} command is already set", id, RESUME ); monitor.notify(); result = command.future; } else if ( type == PAUSE ) { checkState( status == RUNNING, "Pipeline %s cannot be resumed since its status is %s and pending command is %s", id, status, type ); LOGGER.debug( "{}: Completing pending {} command because of new {} command.", id, PAUSE, RESUME ); command.complete(); this.command = null; result = CompletableFuture.completedFuture( TRUE ); } else if ( type == REFRESH ) { LOGGER.debug( "{}: switching pending {} command to {}", id, REFRESH, RESUME ); command = new PipelineReplicaRunnerCommand( RESUME, command.future ); this.command = command; result = command.future; } else { LOGGER.error( "{}: {} failed since there is a pending {} command", id, RESUME, type ); result = new CompletableFuture<>(); command.future.thenRun( () -> result.completeExceptionally( new IllegalStateException( id + ": " + RESUME + " failed since there " + "is a pending " + type + " command" ) ) ); } } else if ( status == RUNNING ) { LOGGER.debug( "{} is already {}", id, RUNNING ); result = CompletableFuture.completedFuture( TRUE ); } else if ( status == PAUSED ) { LOGGER.debug( "{}: {} command is set", id, RESUME ); command = new PipelineReplicaRunnerCommand( RESUME ); this.command = command; result = command.future; } else { LOGGER.error( "{}: {} failed since status is {}", id, RESUME, status ); result = new CompletableFuture<>(); result.completeExceptionally( new IllegalStateException( id + ": " + RESUME + " failed since status is " + status ) ); } } return result; } PipelineReplicaRunner( final JokerConfig config,
                            final PipelineReplica pipeline,
                            final Supervisor supervisor,
                            final DownstreamCollector downstreamCollector ); }
PipelineReplicaRunner implements Runnable { CompletableFuture<Boolean> resume () { final CompletableFuture<Boolean> result; synchronized ( monitor ) { PipelineReplicaRunnerCommand command = this.command; final PipelineReplicaRunnerStatus status = this.status; if ( command != null ) { final PipelineReplicaRunnerCommandType type = command.type; if ( type == RESUME ) { checkState( status == PAUSED, "%s: cannot be resumed since its status is %s and pending command is %s", id, status, type ); LOGGER.debug( "{}: {} command is already set", id, RESUME ); monitor.notify(); result = command.future; } else if ( type == PAUSE ) { checkState( status == RUNNING, "Pipeline %s cannot be resumed since its status is %s and pending command is %s", id, status, type ); LOGGER.debug( "{}: Completing pending {} command because of new {} command.", id, PAUSE, RESUME ); command.complete(); this.command = null; result = CompletableFuture.completedFuture( TRUE ); } else if ( type == REFRESH ) { LOGGER.debug( "{}: switching pending {} command to {}", id, REFRESH, RESUME ); command = new PipelineReplicaRunnerCommand( RESUME, command.future ); this.command = command; result = command.future; } else { LOGGER.error( "{}: {} failed since there is a pending {} command", id, RESUME, type ); result = new CompletableFuture<>(); command.future.thenRun( () -> result.completeExceptionally( new IllegalStateException( id + ": " + RESUME + " failed since there " + "is a pending " + type + " command" ) ) ); } } else if ( status == RUNNING ) { LOGGER.debug( "{} is already {}", id, RUNNING ); result = CompletableFuture.completedFuture( TRUE ); } else if ( status == PAUSED ) { LOGGER.debug( "{}: {} command is set", id, RESUME ); command = new PipelineReplicaRunnerCommand( RESUME ); this.command = command; result = command.future; } else { LOGGER.error( "{}: {} failed since status is {}", id, RESUME, status ); result = new CompletableFuture<>(); result.completeExceptionally( new IllegalStateException( id + ": " + RESUME + " failed since status is " + status ) ); } } return result; } PipelineReplicaRunner( final JokerConfig config,
                            final PipelineReplica pipeline,
                            final Supervisor supervisor,
                            final DownstreamCollector downstreamCollector ); PipelineReplicaRunnerStatus getStatus(); void run(); }
PipelineReplicaRunner implements Runnable { CompletableFuture<Boolean> resume () { final CompletableFuture<Boolean> result; synchronized ( monitor ) { PipelineReplicaRunnerCommand command = this.command; final PipelineReplicaRunnerStatus status = this.status; if ( command != null ) { final PipelineReplicaRunnerCommandType type = command.type; if ( type == RESUME ) { checkState( status == PAUSED, "%s: cannot be resumed since its status is %s and pending command is %s", id, status, type ); LOGGER.debug( "{}: {} command is already set", id, RESUME ); monitor.notify(); result = command.future; } else if ( type == PAUSE ) { checkState( status == RUNNING, "Pipeline %s cannot be resumed since its status is %s and pending command is %s", id, status, type ); LOGGER.debug( "{}: Completing pending {} command because of new {} command.", id, PAUSE, RESUME ); command.complete(); this.command = null; result = CompletableFuture.completedFuture( TRUE ); } else if ( type == REFRESH ) { LOGGER.debug( "{}: switching pending {} command to {}", id, REFRESH, RESUME ); command = new PipelineReplicaRunnerCommand( RESUME, command.future ); this.command = command; result = command.future; } else { LOGGER.error( "{}: {} failed since there is a pending {} command", id, RESUME, type ); result = new CompletableFuture<>(); command.future.thenRun( () -> result.completeExceptionally( new IllegalStateException( id + ": " + RESUME + " failed since there " + "is a pending " + type + " command" ) ) ); } } else if ( status == RUNNING ) { LOGGER.debug( "{} is already {}", id, RUNNING ); result = CompletableFuture.completedFuture( TRUE ); } else if ( status == PAUSED ) { LOGGER.debug( "{}: {} command is set", id, RESUME ); command = new PipelineReplicaRunnerCommand( RESUME ); this.command = command; result = command.future; } else { LOGGER.error( "{}: {} failed since status is {}", id, RESUME, status ); result = new CompletableFuture<>(); result.completeExceptionally( new IllegalStateException( id + ": " + RESUME + " failed since status is " + status ) ); } } return result; } PipelineReplicaRunner( final JokerConfig config,
                            final PipelineReplica pipeline,
                            final Supervisor supervisor,
                            final DownstreamCollector downstreamCollector ); PipelineReplicaRunnerStatus getStatus(); void run();  }