Java 类com.google.common.eventbus.AsyncEventBus 实例源码

private void registerResources(Environment environment) {
    EventStore eventStore = new InMemoryEventStore();
    EventBus eventBus = new AsyncEventBus(newSingleThreadExecutor());

    // domain model
    AccountService accountService = new AccountService(eventStore, eventBus);
    environment.jersey().register(new AccountsResource(accountService));
    environment.jersey().register(new AccountResource(accountService));
    environment.jersey().register(new DepositsResource(accountService));
    environment.jersey().register(new WithdrawalsResource(accountService));

    ClientService clientService = new ClientService(eventStore);
    environment.jersey().register(new ClientsResource(clientService));
    environment.jersey().register(new ClientResource(clientService));

    // read model
    TransactionsRepository transactionsRepository = new InMemoryTransactionsRepository();
    eventBus.register(new TransactionsListener(transactionsRepository));
    environment.jersey().register(new AccountTransactionsResource(transactionsRepository));

    AccountsRepository accountsRepository = new InMemoryAccountsRepository();
    eventBus.register(new AccountsListener(accountsRepository));
    environment.jersey().register(new ClientAccountsResource(accountsRepository));
}

@Provides
@Singleton
EventBus provideEventBus(@AsyncExecutor Executor executor, StatsProvider statsProvider) {
  final AtomicLong subscriberExceptions = statsProvider.makeCounter(EXCEPTIONS_STAT);
  EventBus eventBus = new AsyncEventBus(
      executor,
      (exception, context) -> {
        subscriberExceptions.incrementAndGet();
        log.error(
            "Failed to dispatch event to " + context.getSubscriberMethod() + ": " + exception,
            exception);
      }
  );

  eventBus.register(new DeadEventHandler());
  return eventBus;
}

public static void main(String[] args) {
    ExecutorService executor = Executors.newFixedThreadPool(5);
    EventBus eventBus = new AsyncEventBus("alert-bus", executor);

    Map<String, JsonElement> standardMetadata = ImmutableMap.of("key", JsonUtils.toJson("value"));
    Alerter alerter = new HttpAlerter(new HttpAlerterConfig(Arrays.asList(DESTINATION_URL), ".*", null),
            standardMetadata);

    eventBus.register(alerter);
    // should NOT be sent (doesn't match severity filter)
    eventBus.post(new Alert("/topic", AlertSeverity.INFO, UtcTime.now(), "hello info", null));
    // should be sent (matches severity filter)
    eventBus.post(new Alert("/topic", AlertSeverity.WARN, UtcTime.now(), "hello warning", null));
    eventBus.unregister(alerter);

    executor.shutdownNow();
}

@Override
protected void configure() {
    bind(ActionRepository.class).to(CassandraActionRepository.class);
    bind(PlayerRepository.class).to(CassandraPlayerRepository.class);
    bind(BadgeRepository.class).to(CassandraBadgeRepository.class);
    bind(EventRepository.class).to(CassandraEventRepository.class);
    bind(EventTypeRepository.class).to(CassandraEventTypeRepository.class);

    this.eventBus =  new AsyncEventBus(java.util.concurrent.Executors.newCachedThreadPool());
    bind(EventBus.class).toInstance(eventBus);

    bindListener(Matchers.any(), new TypeListener() {
        public <I> void hear(TypeLiteral<I> typeLiteral, TypeEncounter<I> typeEncounter) {
            typeEncounter.register(new InjectionListener<I>() {
                public void afterInjection(I i) {
                    eventBus.register(i);
                }
            });
        }
    });
}

/**
 * Creates a {@link KubernetesCloudPool}.
 *
 * @param apiServerClient
 *            A client that can be configured to execute (authenticated)
 *            HTTP requests against the REST API of a certain Kubernetes API
 *            server.
 * @param executor
 *            Executor used to schedule periodical tasks.
 * @param eventBus
 *            {@link EventBus} used to push {@link Alert}s. May be
 *            <code>null</code>, in which case a default one is created.
 */
public KubernetesCloudPool(ApiServerClient apiServerClient, ScheduledExecutorService executor, EventBus eventBus) {
    checkArgument(apiServerClient != null, "apiServerClient cannot be null");
    checkArgument(executor != null, "executor cannot be null");

    this.apiServerClient = apiServerClient;
    this.executor = executor;
    this.eventBus = eventBus != null ? eventBus : new AsyncEventBus(executor);

    this.alerter = new MultiplexingAlerter();
    this.eventBus.register(this.alerter);

    this.podPool = null;
    this.config = null;
    this.started = false;
}

synchronized public void initialize() throws Exception {
    if (executor != null) {
        throw new IllegalStateException("Already initialized");
    }
    executor = Executors.newScheduledThreadPool(1);
    eventBus = new AsyncEventBus(executor);
}

@SuppressWarnings("rawtypes")
    public void setEventBus(AsyncEventBus eventBus) {
//      if (callable instanceof PersistenceRequest) {
//          PersistenceRequest<?> request = (PersistenceRequest)callable;
//          request.setEventBus(eventBus);
//      }
    }

SessionEventBus(final Session session) {
  this.session = session;
  this.executorService = Executors.newCachedThreadPool();

  this.eventBus =
      new AsyncEventBus(
          executorService,
          (throwable, subscriberExceptionContext) -> {
            if (!(throwable instanceof RejectedExecutionException)) {
              log.error(throwable.getMessage(), throwable);
            }
          });
}

public AbstractBot() {
    eventBus = new AsyncEventBus(
            ThreadUtil.newCachedThreadPool("event-dispatcher-%d"),
            (exception, context) -> {
                App.logger.catching(exception);
                Watcher.logException();
            }
    );

    eventBus.register(new MessageEventHandler());
    channels = new ConcurrentHashMap<>();
    channelManager = new ChannelManager(channels);
}

@Override
protected EventBus _createEventBusInstance() {
    ExecutorService execService = _executorServiceManager.getExecutorService();
    _eventBusInstance = new AsyncEventBus("R01 ASYNC EventBus",
                                          execService);
    return _eventBusInstance;
}

@Autowired
public EventBusSubscriberBeanPostProcessor(final EventBus eventBus, final AsyncEventBus asyncEventBus) {
    Assert.notNull(eventBus, "EventBus should not be null");
    Assert.notNull(asyncEventBus, "AsyncEventBus should not be null");
    this.eventBus = eventBus;
    this.asyncEventBus = asyncEventBus;
}

public GoogleContainerEngineCloudPool(ContainerClusterClient gkeClient, EventBus eventBus,
        ScheduledExecutorService executor) {
    this.client = gkeClient;
    this.executor = executor;
    this.eventBus = Optional.ofNullable(eventBus).orElse(new AsyncEventBus(this.executor));

    this.alerter = new MultiplexingAlerter();
    this.eventBus.register(this.alerter);

    this.desiredSize = null;
}

@VisibleForTesting
BuckEventBus(Clock clock,
             ExecutorService executorService,
             BuildId buildId,
             int shutdownTimeoutMillis) {
  this.clock = Preconditions.checkNotNull(clock);
  this.executorService = Preconditions.checkNotNull(executorService);
  this.eventBus = new AsyncEventBus("buck-build-events", executorService);
  this.threadIdSupplier = DEFAULT_THREAD_ID_SUPPLIER;
  this.buildId = Preconditions.checkNotNull(buildId);
  this.shutdownTimeoutMillis = shutdownTimeoutMillis;
}

/**
 * Creates a new asynchronous event bus.
 * 
 * @return Async event bus
 */
private EventBus newAsyncEventBus() {
    if (EnvironmentUtil.isUnitTest()) {
        return new EventBus();
    } else {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());
        asyncExecutorList.add(executor);
        return new AsyncEventBus(executor);
    }
}

public JobEventBus(final JobEventConfiguration jobEventConfig) {
    this.jobEventConfig = jobEventConfig;
    executorServiceObject = new ExecutorServiceObject("job-event", Runtime.getRuntime().availableProcessors() * 2);
    eventBus = new AsyncEventBus(executorServiceObject.createExecutorService());
    register();
}

@Bean
@ConditionalOnMissingBean(EventBus.class)
public EventBus eventBus() {
    return new AsyncEventBus(this.getClass().getSimpleName(), Executors.newCachedThreadPool());
}

public static void create(Executor executor) {
    eventBus = new AsyncEventBus(executor, (exception, context)
            -> LOGGER.error("event bus subscriber ex", exception));
}

public ScannerViewModel(String scanDirectory, AsyncEventBus eventBus, TagListVM tagListVm) {
    this.scanDirectory = scanDirectory;
    this.tagListVm = tagListVm;
    eventBus.register(new DirWatcherEventHandler());
}

public BackendThreadPoolExecutor(AsyncEventBus eventBus) {
    super(2, 6, 60, TimeUnit.SECONDS, new PriorityBlockingQueue<Runnable>());
    this.eventBus = eventBus;
}

public BackendManager(AsyncEventBus eventBus) {
    executor = new BackendThreadPoolExecutor(eventBus);
    executor.prestartCoreThread();
    executor.setKeepAliveTime(60, TimeUnit.SECONDS);
}

public FileAlterationHandler(AsyncEventBus eventBus) {
    this.eventBus = eventBus;
}

public DirWatcherManager(AsyncEventBus eventBus) {
    this(eventBus, null);
}

public DirWatcherManager(AsyncEventBus eventBus, String directory) {
    this.dirWatcher = new DirWatcherThread(new FileAlterationHandler(eventBus));
    thread = new Thread(dirWatcher);

    thread.start();
}

public static void main(String[] args) {
    try {
        LanguageController languageController = new LanguageController(); // blehhhhhh
        Message.init(languageController);

        GraphicsProvider launcher = getGraphicsProvider().newInstance();

        launcher.startSplash();
        launcher.updateSplash(Message.STARTUP_PREPARING_ENVIRONMENT);

        Field defaultCharset = Charset.class.getDeclaredField("defaultCharset");
        defaultCharset.setAccessible(true);
        defaultCharset.set(null, StandardCharsets.UTF_8);
        if (!Charset.defaultCharset().equals(StandardCharsets.UTF_8))
            throw new RuntimeException("Charset: " + Charset.defaultCharset());
        if (!Constants.DATA_DIR.exists() && !Constants.DATA_DIR.mkdirs())
            throw new RuntimeException("Could not create data directory");
        if (!Constants.ADDONS_DIR.exists() && !Constants.ADDONS_DIR.mkdirs())
            throw new RuntimeException("Could not create addons directory");
        if (Constants.DATA_DIR.isFile())
            throw new RuntimeException("Data directory is file");
        if (Constants.ADDONS_DIR.isFile())
            throw new RuntimeException("Addons directory is file");

        EventBus eventBus = new AsyncEventBus(Executors.newCachedThreadPool());

        Configuration configuration = loadConfiguration();
        Class<? extends UserInterfaceController> uiController = getUIControllerImpl();

        Injector mainInjector = Guice.createInjector(
                new AbstractModule() {
                    @Override
                    protected void configure() {
                        bind(MessageHandler.class).to(launcher.getMessageHandlerImpl());
                        bind(UserInterfaceController.class).to(uiController);
                        bind(Configuration.class).toInstance(configuration);
                        bind(EventBus.class).toInstance(eventBus);
                    }
                }
        );

        mainInjector.getInstance(UserInterfaceController.class).initialize();

        launcher.updateSplash(Message.STARTUP_LOADING_GRAPHICS);
        launcher.prepare(mainInjector);

        launcher.updateSplash(Message.STARTUP_DONE);
        launcher.start();

        mainInjector.getInstance(PathController.class).reload();
        mainInjector.getInstance(UpdateController.class).doUpdate();
        handleCommandLine(args, mainInjector);
    } catch (Throwable t) {
        displayError(t);
        System.exit(1);
    }
}

public EventDispatcher(Integer threadCount) {
    eventBus = new AsyncEventBus(Executors.newFixedThreadPool(threadCount));
    Logs.info("event dispatcher has started!");
}

public OpcUaServer(OpcUaServerConfig config) {
    this.config = config;

    stackServer = new UaTcpStackServer(config);

    stackServer.addServiceSet((AttributeServiceSet) sessionManager);
    stackServer.addServiceSet((AttributeHistoryServiceSet) sessionManager);
    stackServer.addServiceSet((MethodServiceSet) sessionManager);
    stackServer.addServiceSet((MonitoredItemServiceSet) sessionManager);
    stackServer.addServiceSet((NodeManagementServiceSet) sessionManager);
    stackServer.addServiceSet((SessionServiceSet) sessionManager);
    stackServer.addServiceSet((SubscriptionServiceSet) sessionManager);
    stackServer.addServiceSet((ViewServiceSet) sessionManager);

    ObjectTypeManagerInitializer.initialize(objectTypeManager);
    VariableTypeManagerInitializer.initialize(variableTypeManager);

    namespaceManager.addNamespace(uaNamespace = new OpcUaNamespace(this));

    vendorNamespace = namespaceManager.registerAndAdd(
        config.getApplicationUri(),
        index -> new VendorNamespace(OpcUaServer.this, config.getApplicationUri()));

    serverTable.addUri(stackServer.getApplicationDescription().getApplicationUri());

    for (ReferenceType referenceType : BuiltinReferenceType.values()) {
        referenceTypes.put(referenceType.getNodeId(), referenceType);
    }

    for (String bindAddress : config.getBindAddresses()) {
        Set<String> hostnames = Sets.newHashSet(config.getEndpointAddresses());

        for (String hostname : hostnames) {
            for (SecurityPolicy securityPolicy : config.getSecurityPolicies()) {
                MessageSecurityMode messageSecurity = securityPolicy == SecurityPolicy.None ?
                    MessageSecurityMode.None : MessageSecurityMode.SignAndEncrypt;

                String endpointUrl = endpointUrl(hostname, config.getBindPort(), config.getServerName());

                Set<X509Certificate> certificates = config.getCertificateManager().getCertificates();

                if (certificates.isEmpty()) {
                    logger.info("Binding endpoint {} to {} [{}/{}]",
                        endpointUrl, bindAddress, securityPolicy, messageSecurity);

                    stackServer.addEndpoint(endpointUrl, bindAddress, null, securityPolicy, messageSecurity);
                } else {
                    for (X509Certificate certificate : certificates) {
                        logger.info("Binding endpoint {} to {} [{}/{}]",
                            endpointUrl, bindAddress, securityPolicy, messageSecurity);

                        stackServer.addEndpoint(
                            endpointUrl, bindAddress, certificate, securityPolicy, messageSecurity);
                    }
                }
            }
        }
    }

    eventBus = new AsyncEventBus("server", stackServer.getExecutorService());

    logger.info("eclipse milo opc-ua stack version: {}", Stack.VERSION);
    logger.info("eclipse milo opc-ua sdk version: {}", SDK_VERSION);
}

public ManagedEventBus(final String name) {
    checkNotNull(name, "name is null");
    this.executor = Executors.newScheduledThreadPool(10,
            new ThreadFactoryBuilder().setDaemon(true).setNameFormat("eventbus-" + name + "-%d").build());
    this.eventBus = new AsyncEventBus(executor, new EventBusExceptionHandler(name));
}

/**
 * @return an asynchronous event bus to publish and retrieve events.
 */
@Bean
EventBus eventBus() {
    return new AsyncEventBus(Executors.newFixedThreadPool(4));
}

public SlowProcessSubscriber(AsyncEventBus asyncEventBus, CountDownLatch doneSignal) {
    asyncEventBus.register(this);
    this.doneSignal = doneSignal;
}

public static void main(String[] args) {
    try {
        LanguageController languageController = new LanguageController(); // blehhhhhh
        Message.init(languageController);

        GraphicsProvider launcher = getGraphicsProvider().newInstance();

        launcher.startSplash();
        launcher.updateSplash(Message.STARTUP_PREPARING_ENVIRONMENT);

        Field defaultCharset = Charset.class.getDeclaredField("defaultCharset");
        defaultCharset.setAccessible(true);
        defaultCharset.set(null, StandardCharsets.UTF_8);
        if (!Charset.defaultCharset().equals(StandardCharsets.UTF_8))
            throw new RuntimeException("Charset: " + Charset.defaultCharset());
        if (!Constants.DATA_DIR.exists() && !Constants.DATA_DIR.mkdirs())
            throw new RuntimeException("Could not create data directory");
        if (!Constants.ADDONS_DIR.exists() && !Constants.ADDONS_DIR.mkdirs())
            throw new RuntimeException("Could not create addons directory");
        if (Constants.DATA_DIR.isFile())
            throw new RuntimeException("Data directory is file");
        if (Constants.ADDONS_DIR.isFile())
            throw new RuntimeException("Addons directory is file");

        EventBus eventBus = new AsyncEventBus(Executors.newCachedThreadPool());

        Configuration configuration = loadConfiguration();
        Class<? extends UserInterfaceController> uiController = getUIControllerImpl();

        Injector mainInjector = Guice.createInjector(
                new AbstractModule() {
                    @Override
                    protected void configure() {
                        bind(MessageHandler.class).to(launcher.getMessageHandlerImpl());
                        bind(UserInterfaceController.class).to(uiController);
                        bind(Configuration.class).toInstance(configuration);
                        bind(EventBus.class).toInstance(eventBus);
                    }
                }
        );

        mainInjector.getInstance(UserInterfaceController.class).initialize();

        launcher.updateSplash(Message.STARTUP_LOADING_GRAPHICS);
        launcher.prepare(mainInjector);

        launcher.updateSplash(Message.STARTUP_DONE);
        launcher.start();

        mainInjector.getInstance(PathController.class).reload();
        mainInjector.getInstance(UpdateController.class).doUpdate();
        handleCommandLine(args, mainInjector);
    } catch (Throwable t) {
        displayError(t);
        System.exit(1);
    }
}

@Inject
GripPlatform(EventBus eventBus) {
  checkArgument(!(eventBus instanceof AsyncEventBus), "This class has not been tested to work "
      + "with the AsyncEventBus");
  this.eventBus = eventBus;
}

EventBusSupportImpl(final EventBus eventBus, final AsyncEventBus asyncEventBus) {
    Assert.notNull(eventBus, "EventBus should not be null");
    Assert.notNull(asyncEventBus, "AsyncEventBus should not be null");
    this.eventBus = eventBus;
    this.asyncEventBus = asyncEventBus;
}

@Autowired
public SimpleAsyncEventSender(final AsyncEventBus asyncEventSender) {
    this.asyncEventBus = asyncEventSender;
}

@Autowired
public MultiEventBusSender(final AsyncEventBus asyncEventBus, final EventBus eventBus) {
    this.asyncEventBus = asyncEventBus;
    this.eventBus = eventBus;
}

@Bean
public EventBus eventBus() {
  return new AsyncEventBus(executor);
}

@SuppressWarnings("unchecked")
public void handle(Deployment deployment)
{
    NodeManagerFactory.getManagerReference();

    try
    {

        WorkflowDescription workflowDescription = deploymentService.createWorkflowFor(deployment);
        String thisHostname = System.getProperty("org.excalibur.instance.hostname");

        VirtualMachine localNode = instanceService_.getInstanceByName(thisHostname);
        checkState(localNode != null);

        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("event-bus-for-workflow-" + workflowDescription.getName() + "-%d")
                .build();

        int numberOfActivities = workflowDescription.getNumberOfActivities();

        LOG.debug("workflow [{}] has [{}] activities [{}]", workflowDescription.getName(), numberOfActivities);

        ExecutorService eventBusExecutor = DynamicExecutors.newScalingThreadPool(1, numberOfActivities, 5, TimeUnit.MINUTES, threadFactory);
        EventBus bus = new AsyncEventBus(eventBusExecutor);

        final Workflow workflow = new WorkflowBuilder().description(workflowDescription).eventBus(bus).build();
        final WorkflowExecutionStrategy strategy = new WorkflowExecutionStrategy(workflow);

        WorkflowContext context = new WorkflowContextImpl(workflow, workflowRepository_, strategy);
        context.setWorkflowCoordinator(localNode);

        context.registerExecutors(eventBusExecutor);

        final WorkflowExecutor executor = new WorkflowExecutor(context, taskRepository_, userRepository_, regionRepository_, localNode);
        List<ActivityExecutionContext> executionContexts = executor.execute();

        for (ActivityExecutionContext executionContext : executionContexts)
        {
            for (TaskType<?> task : executionContext.getTasks())
            {
                TaskResult<Instances> result = (TaskResult<Instances>) task.getResult();

                if (result != null && TaskState.SUCCESS.equals(result.getTaskState()))
                {
                    LOG.debug("Task [{}] executed successfully in [{}] ms",
                            TimeUnit.MILLISECONDS.toMillis(result.getFinishTime() - result.getStartTime()));

                    newInstanceTemplate_.convertAndSend(result.getResult());
                }
            }
        }

        long elapsedTime = workflowDescription.getCreatedIn() != null ? System.currentTimeMillis() - workflowDescription.getCreatedIn().getTime() : 0;
        LOG.debug("Finished the workflow [{}] in [{}] seconds", workflowDescription.getName(), TimeUnit.MILLISECONDS.toSeconds(elapsedTime));
    }
    catch (Exception exception)
    {
        String text = DeploymentUtils.marshalQuietly(deployment);
        LOG.error("Error on executing the deployment [{}]. Error message [{}]", text, exception.getMessage(), exception);
        AnyThrow.throwUncheked(exception);
    }
}

public static AsyncEventBus createAsyncEventBus(ListeningExecutorService executor, DeadEventLoggingHandler deadEventLoggingHandler)
{
    return createAsyncEventBus("default-async-event-bus", executor, deadEventLoggingHandler);
}

public static AsyncEventBus createAsyncEventBus(String name, ListeningExecutorService executor, DeadEventLoggingHandler deadEventLoggingHandler)
{
    AsyncEventBus asyncEventBus = new AsyncEventBus(name, executor);
    asyncEventBus.register(deadEventLoggingHandler);
    return asyncEventBus;
}

@Test
public void must_execute_one_workflow() throws InterruptedException, JAXBException
{
    TaskDescription task = new TaskDescription().setExecutable("who").setTypeClass(DummyTask.class.getName());

    WorkflowActivityDescription a = new WorkflowActivityDescription().setLabel("a").setId(1).addTask(task.clone().setId(1));
    WorkflowActivityDescription b = new WorkflowActivityDescription().setLabel("b").setParents("1").setId(2).addTask(task.clone().setId(1));
    WorkflowActivityDescription c = new WorkflowActivityDescription().setLabel("c").setParents("1").setId(3).addTask(task.clone().setId(1));
    WorkflowActivityDescription d = new WorkflowActivityDescription().setLabel("d").setParents("2").setId(4).addTask(task.clone().setId(1));
    WorkflowActivityDescription e = new WorkflowActivityDescription().setLabel("e").setParents("2,3").setId(5).addTask(task.clone().setId(1));
    WorkflowActivityDescription f = new WorkflowActivityDescription().setLabel("f").setParents("4,5").setId(6).addTask(task.clone().setId(1));

    b.addParent(a);
    c.addParent(a);
    d.addParent(b);

    e.addParents(b, c);
    f.addParents(d, e);

    WorkflowDescription workflowDescription = new WorkflowDescription().setName("wk-test").setStartActivityId(a.getId()).addActivities(a, b, c, d, e, f);

    workflowDescription.setId(workflowRepository_.insert(workflowDescription.setUser(user)));

    a.setId(workflowRepository_.insert(a)).setInternalId(a.getId());
    b.setId(workflowRepository_.insert(b)).setInternalId(b.getId());
    c.setId(workflowRepository_.insert(c)).setInternalId(c.getId());
    d.setId(workflowRepository_.insert(d)).setInternalId(d.getId());
    e.setId(workflowRepository_.insert(e)).setInternalId(e.getId());
    f.setId(workflowRepository_.insert(f)).setInternalId(f.getId());

    a.getTasks().get(0).setId(taskRepository_.insert(a.getTasks().get(0)));
    b.getTasks().get(0).setId(taskRepository_.insert(b.getTasks().get(0)));
    c.getTasks().get(0).setId(taskRepository_.insert(c.getTasks().get(0)));
    d.getTasks().get(0).setId(taskRepository_.insert(d.getTasks().get(0)));
    e.getTasks().get(0).setId(taskRepository_.insert(e.getTasks().get(0)));
    f.getTasks().get(0).setId(taskRepository_.insert(f.getTasks().get(0)));


    JAXBContextFactory<WorkflowDescription> factory = new JAXBContextFactory<WorkflowDescription>(WorkflowDescription.class);
    assertNotNull(factory.marshal(workflowDescription));

    ThreadFactory threadFactory2 = new ThreadFactoryBuilder()
            .setNameFormat("event-bus-for-workflow-" + workflowDescription.getName() + "-%d")
            .build();

    ExecutorService executorService = DynamicExecutors.newScalingThreadPool(6, 100, 3, TimeUnit.MINUTES, threadFactory2);
    EventBus bus = new AsyncEventBus(executorService);

    final Workflow workflow = new WorkflowBuilder().description(workflowDescription).eventBus(bus).build();
    final WorkflowExecutionStrategy strategy = new WorkflowExecutionStrategy(workflow);

    WorkflowContext context = new WorkflowContextImpl(null, workflow, workflowRepository_, strategy);
    context.setWorkflowCoordinator(localNode);
    context.registerExecutors(executorService);

    final WorkflowExecutor executor = new WorkflowExecutor(context, taskRepository_, userRepository_, regionRepository_, localNode);
    executor.execute();
    assertTrue(workflow.isFinished());
}