@Test public void get_eventBus_from_engine() { final CamundaEventBus eventBus = new CamundaEventBus(); ProcessEngine engine = new ReactorProcessEngineConfiguration(eventBus).buildProcessEngine(); try { assertThat(CamundaReactor.eventBus(engine)).isEqualTo(eventBus); } finally { engine.close(); } }
public static CamundaEventBus eventBus(final ProcessEngine processEngine) { final ProcessEngineConfigurationImpl configuration = (ProcessEngineConfigurationImpl) processEngine.getProcessEngineConfiguration(); final List<ProcessEnginePlugin> plugins = Optional.ofNullable(configuration.getProcessEnginePlugins()).orElse(Collections.EMPTY_LIST); final Function<List<ProcessEnginePlugin>, Optional<CamundaEventBus>> filterReactorPlugin = l -> l.stream() .filter(plugin -> plugin instanceof ReactorProcessEnginePlugin) .map(ReactorProcessEnginePlugin.class::cast) .map(ReactorProcessEnginePlugin::getEventBus) .findFirst(); Optional<CamundaEventBus> reactorProcessEnginePlugin = filterReactorPlugin.apply(plugins); if (reactorProcessEnginePlugin.isPresent()) { return reactorProcessEnginePlugin.get(); } return plugins.stream() .filter(plugin -> plugin instanceof CompositeProcessEnginePlugin) .map(CompositeProcessEnginePlugin.class::cast) .map(CompositeProcessEnginePlugin::getPlugins) .map(filterReactorPlugin) .flatMap(o -> o.map(Stream::of).orElseGet(Stream::empty)) .findFirst() .orElseThrow(illegalState("No eventBus found. Make sure the Reactor plugin is configured correctly.")); }
CamundaReactor { public static CamundaEventBus eventBus(final ProcessEngine processEngine) { final ProcessEngineConfigurationImpl configuration = (ProcessEngineConfigurationImpl) processEngine.getProcessEngineConfiguration(); final List<ProcessEnginePlugin> plugins = Optional.ofNullable(configuration.getProcessEnginePlugins()).orElse(Collections.EMPTY_LIST); final Function<List<ProcessEnginePlugin>, Optional<CamundaEventBus>> filterReactorPlugin = l -> l.stream() .filter(plugin -> plugin instanceof ReactorProcessEnginePlugin) .map(ReactorProcessEnginePlugin.class::cast) .map(ReactorProcessEnginePlugin::getEventBus) .findFirst(); Optional<CamundaEventBus> reactorProcessEnginePlugin = filterReactorPlugin.apply(plugins); if (reactorProcessEnginePlugin.isPresent()) { return reactorProcessEnginePlugin.get(); } return plugins.stream() .filter(plugin -> plugin instanceof CompositeProcessEnginePlugin) .map(CompositeProcessEnginePlugin.class::cast) .map(CompositeProcessEnginePlugin::getPlugins) .map(filterReactorPlugin) .flatMap(o -> o.map(Stream::of).orElseGet(Stream::empty)) .findFirst() .orElseThrow(illegalState("No eventBus found. Make sure the Reactor plugin is configured correctly.")); } }
CamundaReactor { public static CamundaEventBus eventBus(final ProcessEngine processEngine) { final ProcessEngineConfigurationImpl configuration = (ProcessEngineConfigurationImpl) processEngine.getProcessEngineConfiguration(); final List<ProcessEnginePlugin> plugins = Optional.ofNullable(configuration.getProcessEnginePlugins()).orElse(Collections.EMPTY_LIST); final Function<List<ProcessEnginePlugin>, Optional<CamundaEventBus>> filterReactorPlugin = l -> l.stream() .filter(plugin -> plugin instanceof ReactorProcessEnginePlugin) .map(ReactorProcessEnginePlugin.class::cast) .map(ReactorProcessEnginePlugin::getEventBus) .findFirst(); Optional<CamundaEventBus> reactorProcessEnginePlugin = filterReactorPlugin.apply(plugins); if (reactorProcessEnginePlugin.isPresent()) { return reactorProcessEnginePlugin.get(); } return plugins.stream() .filter(plugin -> plugin instanceof CompositeProcessEnginePlugin) .map(CompositeProcessEnginePlugin.class::cast) .map(CompositeProcessEnginePlugin::getPlugins) .map(filterReactorPlugin) .flatMap(o -> o.map(Stream::of).orElseGet(Stream::empty)) .findFirst() .orElseThrow(illegalState("No eventBus found. Make sure the Reactor plugin is configured correctly.")); } private  CamundaReactor(); }
CamundaReactor { public static CamundaEventBus eventBus(final ProcessEngine processEngine) { final ProcessEngineConfigurationImpl configuration = (ProcessEngineConfigurationImpl) processEngine.getProcessEngineConfiguration(); final List<ProcessEnginePlugin> plugins = Optional.ofNullable(configuration.getProcessEnginePlugins()).orElse(Collections.EMPTY_LIST); final Function<List<ProcessEnginePlugin>, Optional<CamundaEventBus>> filterReactorPlugin = l -> l.stream() .filter(plugin -> plugin instanceof ReactorProcessEnginePlugin) .map(ReactorProcessEnginePlugin.class::cast) .map(ReactorProcessEnginePlugin::getEventBus) .findFirst(); Optional<CamundaEventBus> reactorProcessEnginePlugin = filterReactorPlugin.apply(plugins); if (reactorProcessEnginePlugin.isPresent()) { return reactorProcessEnginePlugin.get(); } return plugins.stream() .filter(plugin -> plugin instanceof CompositeProcessEnginePlugin) .map(CompositeProcessEnginePlugin.class::cast) .map(CompositeProcessEnginePlugin::getPlugins) .map(filterReactorPlugin) .flatMap(o -> o.map(Stream::of).orElseGet(Stream::empty)) .findFirst() .orElseThrow(illegalState("No eventBus found. Make sure the Reactor plugin is configured correctly.")); } private  CamundaReactor(); static DelegateTaskEvent wrap(final DelegateTask delegateTask); static DelegateExecutionEvent wrap(final DelegateExecution delegateExecution); static DelegateCaseExecutionEvent wrap(final DelegateCaseExecution delegateCaseExecution); static SelectorBuilder selector(); static CamundaEventBus eventBus(final ProcessEngine processEngine); static CamundaEventBus eventBus(); static ReactorProcessEnginePlugin plugin(); static ReactorProcessEnginePlugin plugin(final CamundaEventBus camundaEventBus); }
CamundaReactor { public static CamundaEventBus eventBus(final ProcessEngine processEngine) { final ProcessEngineConfigurationImpl configuration = (ProcessEngineConfigurationImpl) processEngine.getProcessEngineConfiguration(); final List<ProcessEnginePlugin> plugins = Optional.ofNullable(configuration.getProcessEnginePlugins()).orElse(Collections.EMPTY_LIST); final Function<List<ProcessEnginePlugin>, Optional<CamundaEventBus>> filterReactorPlugin = l -> l.stream() .filter(plugin -> plugin instanceof ReactorProcessEnginePlugin) .map(ReactorProcessEnginePlugin.class::cast) .map(ReactorProcessEnginePlugin::getEventBus) .findFirst(); Optional<CamundaEventBus> reactorProcessEnginePlugin = filterReactorPlugin.apply(plugins); if (reactorProcessEnginePlugin.isPresent()) { return reactorProcessEnginePlugin.get(); } return plugins.stream() .filter(plugin -> plugin instanceof CompositeProcessEnginePlugin) .map(CompositeProcessEnginePlugin.class::cast) .map(CompositeProcessEnginePlugin::getPlugins) .map(filterReactorPlugin) .flatMap(o -> o.map(Stream::of).orElseGet(Stream::empty)) .findFirst() .orElseThrow(illegalState("No eventBus found. Make sure the Reactor plugin is configured correctly.")); } private  CamundaReactor(); static DelegateTaskEvent wrap(final DelegateTask delegateTask); static DelegateExecutionEvent wrap(final DelegateExecution delegateExecution); static DelegateCaseExecutionEvent wrap(final DelegateCaseExecution delegateCaseExecution); static SelectorBuilder selector(); static CamundaEventBus eventBus(final ProcessEngine processEngine); static CamundaEventBus eventBus(); static ReactorProcessEnginePlugin plugin(); static ReactorProcessEnginePlugin plugin(final CamundaEventBus camundaEventBus); static final String CAMUNDA_TOPIC; }