ActivityExecutor implements ActivityController, ParameterMap.Listener, ProgressMeter { public synchronized void startActivity() { logger.info("starting activity " + activity.getAlias() + " for cycles " + activity.getCycleSummary()); try { activity.setRunState(RunState.Starting); activity.initActivity(); } catch (UserException e) { throw e; } catch (Exception e) { this.stoppingException = new RuntimeException("Error initializing activity '" + activity.getAlias() +"': " + e.getMessage(),e); logger.error("error initializing activity '" + activity.getAlias() + "': " + stoppingException); throw stoppingException; } adjustToActivityDef(activity.getActivityDef()); activity.setRunState(RunState.Running); } ActivityExecutor(Activity activity); synchronized void startActivity(); synchronized void stopActivity(); synchronized void forceStopExecutor(int initialMillisToWait); boolean requestStopExecutor(int secondsToWait); @Override synchronized void handleParameterMapUpdate(ParameterMap parameterMap); ActivityDef getActivityDef(); boolean awaitCompletion(int waitTime); boolean awaitFinish(int timeout); String toString(); boolean isRunning(); Activity getActivity(); @Override synchronized double getProgress(); @Override String getProgressDetails(); @Override String getProgressName(); @Override RunState getProgressState(); synchronized void notifyException(Thread t, Throwable e); @Override synchronized void stopActivityWithReasonAsync(String reason); @Override synchronized void stopActivityWithErrorAsync(Throwable throwable);  }
@Test(enabled=true) public void testNewActivityExecutor() { ActivityDef ad = ActivityDef.parseActivityDef("type=diag;alias=test;cycles=1000;"); Optional<ActivityType> activityType = ActivityType.FINDER.get(ad.getActivityType()); Input longSupplier = new AtomicInput(ad); MotorDispenser<?> cmf = getActivityMotorFactory( ad, motorActionDelay(999), longSupplier ); Activity a = new SimpleActivity(ad); InputDispenser idisp = new CoreInputDispenser(a); ActionDispenser adisp = new CoreActionDispenser(a); OutputDispenser tdisp = CoreServices.getOutputDispenser(a).orElse(null); MotorDispenser<?> mdisp = new CoreMotorDispenser(a, idisp, adisp, tdisp); a.setActionDispenserDelegate(adisp); a.setInputDispenserDelegate(idisp); a.setMotorDispenserDelegate(mdisp); ActivityExecutor ae = new ActivityExecutor(a); ad.setThreads(5); ae.startActivity(); int[] speeds = new int[]{1,2000,5,2000,2,2000}; for(int offset=0; offset<speeds.length; offset+=2) { int threadTarget=speeds[offset]; int threadTime = speeds[offset+1]; logger.info("Setting thread level to " + threadTarget + " for " +threadTime + " seconds."); ad.setThreads(threadTarget); try { Thread.sleep(threadTime); } catch (InterruptedException ignored) { } } ad.setThreads(0); }