com.google.inject.PrivateModule#com.google.common.eventbus.EventBus源码实例Demo

public MediaMappingChangeEventHandler() {
    EventBus eventBus = EventBusFactory.getEventBus();
    eventBus.register(new Object() {
        @Subscribe
        public void listener(MediaMappingChangeEvent event) {
            logger.info("Receive an mediaMapping change event with task-id " + event.getTaskId());
            try {
                cleanMediaMapping(event.getTaskId());
                event.getCallback().onCompletion(null, null);
            } catch (Throwable t) {
                logger.error("something goes wrong when clean mediaMapping cache.", t);
                event.getCallback().onCompletion(t, null);
            }
        }
    });
}
 

@Provides
@Singleton
public HeartbeatManager provideHeartBeat(EventBus eventBus, Clock clock) {
  HeartbeatManager heartbeatManager =
      new HeartbeatManager(
          new ConcurrentHashMap<>(),
          MoreExecutors.getExitingScheduledExecutorService(
              (ScheduledThreadPoolExecutor)
                  Executors.newScheduledThreadPool(
                      1,
                      new ThreadFactoryBuilder().setNameFormat("heartbeat-manager-%d").build())),
          eventBus,
          clock,
          heartbeatConfiguration);
  eventBus.register(heartbeatManager);
  return heartbeatManager;
}
 

public void execute(JobExecutionContext context) throws JobExecutionException {
  EventBus eventBus = (EventBus) context.getJobDetail().getJobDataMap().get(EVENT_BUS);
  String collectionName = (String) context.getJobDetail().getJobDataMap().get(COLLECTION_NAME);
  org.slf4j.Logger loggerfc = (org.slf4j.Logger) context.getJobDetail().getJobDataMap().get(LOGGER);
  CollectionUpdateType type = (CollectionUpdateType) context.getTrigger().getJobDataMap().get(UpdateType);
  String source = (String) context.getTrigger().getJobDataMap().get(Source);
  String groupName = context.getTrigger().getKey().getGroup();

  try {
    eventBus.post(new CollectionUpdateEvent(type, collectionName, source));
    fcLogger.debug("CollectionUpdate post event {} on {}", type, collectionName);

  } catch (Throwable e) {
    if (loggerfc != null)
      loggerfc.error("UpdateCollectionJob.execute " + groupName + " failed collection=" + collectionName, e);
  }
}
 

@Test
public void respondsWithCorrectSnapshot() throws IOException {
    EventBus eventBus = new EventBus();
    OriginsInventoryHandler handler = new OriginsInventoryHandler(eventBus);

    Set<Origin> activeOrigins = generateOrigins(3);
    Set<Origin> inactiveOrigins = generateOrigins(4);
    Set<Origin> disabledOrigins = generateOrigins(2);

    eventBus.post(new OriginsSnapshot(APP_ID, pool(activeOrigins), pool(inactiveOrigins), pool(disabledOrigins)));

    HttpResponse response = Mono.from(handler.handle(get("/").build(), requestContext())).block();
    assertThat(response.bodyAs(UTF_8).split("\n").length, is(1));

    Map<Id, OriginsSnapshot> output = deserialiseJson(response.bodyAs(UTF_8));

    assertThat(output.keySet(), contains(APP_ID));

    OriginsSnapshot snapshot = output.get(APP_ID);

    assertThat(snapshot.appId(), is(APP_ID));
    assertThat(snapshot.activeOrigins(), is(activeOrigins));
    assertThat(snapshot.inactiveOrigins(), is(inactiveOrigins));
    assertThat(snapshot.disabledOrigins(), is(disabledOrigins));
}
 

public ScheduledJobConfigurationManager(EventBus eventBus, Config config) {
  super(eventBus, config);

  this.jobSpecs = Maps.newHashMap();
  this.refreshIntervalInSeconds = ConfigUtils.getLong(config, GobblinClusterConfigurationKeys.JOB_SPEC_REFRESH_INTERVAL,
      DEFAULT_JOB_SPEC_REFRESH_INTERVAL);

  this.fetchJobSpecExecutor = Executors.newSingleThreadScheduledExecutor(
      ExecutorsUtils.newThreadFactory(Optional.of(LOGGER), Optional.of("FetchJobSpecExecutor")));

  this.aliasResolver = new ClassAliasResolver<>(SpecConsumer.class);
  try {
    String specConsumerClassName = GobblinClusterConfigurationKeys.DEFAULT_SPEC_CONSUMER_CLASS;
    if (config.hasPath(GobblinClusterConfigurationKeys.SPEC_CONSUMER_CLASS_KEY)) {
      specConsumerClassName = config.getString(GobblinClusterConfigurationKeys.SPEC_CONSUMER_CLASS_KEY);
    }
    LOGGER.info("Using SpecConsumer ClassNameclass name/alias " + specConsumerClassName);
    this._specConsumer = (SpecConsumer) ConstructorUtils
        .invokeConstructor(Class.forName(this.aliasResolver.resolve(specConsumerClassName)), config);
  } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | InstantiationException
      | ClassNotFoundException e) {
    throw new RuntimeException(e);
  }
}
 

@Test
public void testFLUME1854() throws Exception {
  File configFile = new File(baseDir, "flume-conf.properties");
  Files.copy(new File(getClass().getClassLoader()
      .getResource("flume-conf.properties").getFile()), configFile);
  Random random = new Random();
  for (int i = 0; i < 3; i++) {
    EventBus eventBus = new EventBus("test-event-bus");
    PollingPropertiesFileConfigurationProvider configurationProvider =
        new PollingPropertiesFileConfigurationProvider("host1",
            configFile, eventBus, 1);
    List<LifecycleAware> components = Lists.newArrayList();
    components.add(configurationProvider);
    Application application = new Application(components);
    eventBus.register(application);
    application.start();
    Thread.sleep(random.nextInt(10000));
    application.stop();
  }
}
 

@Test
public void prettyPrintsOriginsSnapshot() {
    EventBus eventBus = new EventBus();
    OriginsInventoryHandler handler = new OriginsInventoryHandler(eventBus);

    Set<Origin> disabledOrigins = generateOrigins(2);

    eventBus.post(new OriginsSnapshot(APP_ID, pool(emptySet()), pool(emptySet()), pool(disabledOrigins)));

    HttpResponse response = Mono.from(handler.handle(get("/?pretty=1").build(), requestContext())).block();
    assertThat(body(response).replace("\r\n", "\n"),
            matchesRegex("\\{\n" +
                    "  \"" + APP_ID + "\" : \\{\n" +
                    "    \"appId\" : \"" + APP_ID + "\",\n" +
                    "    \"activeOrigins\" : \\[ ],\n" +
                    "    \"inactiveOrigins\" : \\[ ],\n" +
                    "    \"disabledOrigins\" : \\[ \\{\n" +
                    "      \"id\" : \"origin.\",\n" +
                    "      \"host\" : \"localhost:....\"\n" +
                    "    }, \\{\n" +
                    "      \"id\" : \"origin.\",\n" +
                    "      \"host\" : \"localhost:....\"\n" +
                    "    } ]\n" +
                    "  }\n" +
                    "}"));
}
 

public ValidatorApiHandler(
    final CombinedChainDataClient combinedChainDataClient,
    final SyncStateTracker syncStateTracker,
    final StateTransition stateTransition,
    final BlockFactory blockFactory,
    final AggregatingAttestationPool attestationPool,
    final AttestationManager attestationManager,
    final AttestationTopicSubscriber attestationTopicSubscriber,
    final EventBus eventBus) {
  this.combinedChainDataClient = combinedChainDataClient;
  this.syncStateTracker = syncStateTracker;
  this.stateTransition = stateTransition;
  this.blockFactory = blockFactory;
  this.attestationPool = attestationPool;
  this.attestationManager = attestationManager;
  this.attestationTopicSubscriber = attestationTopicSubscriber;
  this.eventBus = eventBus;
}
 

@BeforeEach
public void setUp() throws Exception {
  final EventBus eventBus1 = new EventBus();
  final RpcEncoding rpcEncoding = getEncoding();

  storageClient1 = MemoryOnlyRecentChainData.create(eventBus1);
  BeaconChainUtil.create(0, storageClient1).initializeStorage();
  final Eth2Network network1 =
      networkFactory
          .builder()
          .rpcEncoding(rpcEncoding)
          .eventBus(eventBus1)
          .recentChainData(storageClient1)
          .startNetwork();
  final Eth2Network network2 =
      networkFactory.builder().rpcEncoding(rpcEncoding).peer(network1).startNetwork();
  peer1 = network2.getPeer(network1.getNodeId()).orElseThrow();
}
 

private void initPreApplicationState() {
    logger.info(Version.getCurrentVersion().toString());
    UI.events = this;

    Thread.currentThread().setUncaughtExceptionHandler((thread, throwable) ->
            logger.error(throwable.getMessage(), throwable));

    TestController.setUI(this, getParameters());
    prefs = TestController.loadApplicationPreferences();

    eventBus = new EventBus();
    if (TestController.isTestMode()) {
        registerTestEvents();
    }
    registerEvent((UnusedStoredReposChangedEventHandler) e -> onRepoOpened());
    registerEvent((UsedReposChangedEventHandler) e -> removeUnusedModelsAndUpdate());

    uiManager = new UIManager(this);
    status = new HTStatusBar(this);

    updateManager = TestController.createUpdateManager();
    updateManager.run();
}
 

@Test
public void testEventBus(){
	EventBus eventBus = new EventBus();
	eventBus.register(new Object(){
		
		@Subscribe
		public void listener(TestEvent event){
			System.out.println("listener1:"+event.name);
		}
	});
	/*eventBus.register(new Object(){
		
		@Subscribe
		public void listener(TestEvent event){
			System.out.println("listener2:"+event.name);
			if(true)
				throw new BaseException("error");
		}
	});
	eventBus.register(new Object(){
		
		@Subscribe
		public void listener(TestEvent event){
			System.out.println("listener3:"+event.name);
			if(true)
				throw new BaseException("error");
		}
	});*/

	eventBus.post(new TestEvent("test"));
	eventBus.post(new TestEvent("test"));
}
 

public void run(FileImport fileImport) {
	EventBus bus = new EventBus();
	bus.register(this);
	fileImport.setEventBus(bus);
	Thread thread = new Thread(fileImport);
	thread.start();
	while (!monitor.isCanceled() && thread.isAlive())
		try {
			thread.join(cancelLookUpTime);
			if (monitor.isCanceled())
				fileImport.cancel();
		} catch (Exception e) {
			log.error("failed to join import thread");
		}
}
 

@Test
// mock ServiceRegistryClientImpl to avoid send request to remote SC
// even there is no any reference to registryClient, DO NOT delete it.
// because what changed is class ServiceRegistryClientImpl
public void testGetRemoteRegistryClient(@Mocked ServiceRegistryClientImpl registryClient) {
  EventBus eventBus = new EventBus();
  ServiceRegistryConfig serviceRegistryConfig = ServiceRegistryConfig.INSTANCE;
  MicroserviceDefinition microserviceDefinition = new MicroserviceDefinition(Collections.emptyList());

  ServiceRegistry serviceRegistry =
      ServiceRegistryFactory.create(eventBus, serviceRegistryConfig, microserviceDefinition);
  serviceRegistry.init();
  ServiceRegistryClient client = serviceRegistry.getServiceRegistryClient();
  Assert.assertTrue(client instanceof ServiceRegistryClientImpl);

  serviceRegistry = ServiceRegistryFactory.create(eventBus,
      serviceRegistryConfig,
      microserviceDefinition);
  Assert.assertTrue(serviceRegistry instanceof RemoteServiceRegistry);

  serviceRegistry = LocalServiceRegistryFactory.createLocal(eventBus, serviceRegistryConfig, microserviceDefinition);
  serviceRegistry.init();
  client = serviceRegistry.getServiceRegistryClient();
  Assert.assertTrue(client instanceof LocalServiceRegistryClientImpl);
  Assert.assertTrue(LocalServiceRegistryFactory.createLocal(eventBus,
      serviceRegistryConfig,
      microserviceDefinition) instanceof LocalServiceRegistry);
}
 

private ThreadLocal<Boolean> getIsDispatching() {
    try {
        final Field field = EventBus.class.getDeclaredField("isDispatching");
        field.setAccessible(true);
        return (ThreadLocal<Boolean>) field.get(this);
    } catch(IllegalAccessException | NoSuchFieldException e) {
        throw new IllegalStateException();
    }
}
 

@Before
public void setUp() {
  clock = Clock.fixed(Instant.now(), ZoneId.of("UTC"));
  tokenManager = new JacksonTokenManager(new ObjectMapper());
  validToken = tokenManager.generateToken(2, "test-cluster", clock.millis() + 1000);
  ObjectMapper mapper = new ObjectMapper();
  transformer = new JacksonJobSpecTransformer(mapper);
  tokenManager = new JacksonTokenManager(mapper);
  scheduleManager =
      spy(new NoOpScheduleManager(64, tokenManager, transformer, clock, new NoOpMetricsImpl()));
  worker =
      new PartitionWorker(
          1,
          asyncClient,
          new EventBus(),
          clock,
          scheduleManager,
          transformer,
          new WorkersConfiguration(1, 10, 1000),
          new NoOpMetricsImpl());
  worker.startExecutor();
  when(scheduleManager.updateStatus(
          any(Schedule.class), eq(JobStatus.SCHEDULED), eq(JobStatus.ACQUIRED)))
      .thenAnswer(
          invocation ->
              CompletableFuture.completedFuture(
                  ((Schedule) invocation.getArguments()[0]).withStatus(JobStatus.ACQUIRED)));
}
 

@Before
public void setUp() throws Exception {

  baseDir = Files.createTempDir();

  configFile = new File(baseDir, TESTFILE.getName());
  Files.copy(TESTFILE, configFile);

  eventBus = new EventBus("test");
  provider =
      new PollingPropertiesFileConfigurationProvider("host1",
          configFile, eventBus, 1);
  provider.start();
  LifecycleController.waitForOneOf(provider, LifecycleState.START_OR_ERROR);
}
 

@BeforeAll
public static void setup() {
  localEventBus = new EventBus();
  recentChainData = MemoryOnlyRecentChainData.create(localEventBus);
  beaconStateInternal = dataStructureUtil.randomBeaconState();

  beaconState = new BeaconState(beaconStateInternal);
  recentChainData.initializeFromGenesis(beaconStateInternal);
  combinedChainDataClient = new CombinedChainDataClient(recentChainData, historicalChainData);
  blockRoot = recentChainData.getBestBlockRoot().orElseThrow();
  slot = recentChainData.getBestSlot();
}
 

@BeforeAll
public static void setup() {
  final EventBus localEventBus = new EventBus();
  final RecentChainData storageClient = MemoryOnlyRecentChainData.create(localEventBus);
  beaconState = dataStructureUtil.randomBeaconState();
  storageClient.initializeFromGenesis(beaconState);
  combinedChainDataClient = new CombinedChainDataClient(storageClient, historicalChainData);
  blockRoot = storageClient.getBestBlockRoot().orElseThrow();
  slot = storageClient.getBlockState(blockRoot).get().getSlot();
  epoch = slot.dividedBy(UnsignedLong.valueOf(SLOTS_PER_EPOCH));
}
 

@Override
public WorkUnitStream getWorkunitStream(SourceState state) {
  int numTasks = state.getPropAsInt(NUM_TASKS_KEY);
  String eventBusId = state.getProp(EventBusPublishingTaskFactory.EVENTBUS_ID_KEY);
  EventBus eventBus = TestingEventBuses.getEventBus(eventBusId);

  return new BasicWorkUnitStream.Builder(new WorkUnitIterator(eventBus, eventBusId, numTasks)).build();
}
 

@Before
public void setup() {
    eventRecieved = new AtomicInteger(
                                      0);
    bus = new EventBus();
    bus.register(this);
    dummyMc = new DummyManifestComparator<>();
    writer = new AsyncDataWriter<>(
                                   ex, dummyMc, bus);
}
 

@Test
public void testGetPackageUsesListener() throws Exception {
  scratch.file("pkg/BUILD", "thisisanerror");
  EventCollector customEventCollector = new EventCollector(EventKind.ERRORS);
  Package pkg = skyframeExecutor.getPackageManager().getPackage(
      new Reporter(new EventBus(), customEventCollector),
      PackageIdentifier.createInMainRepo("pkg"));
  assertThat(pkg.containsErrors()).isTrue();
  MoreAsserts.assertContainsEvent(customEventCollector, "name 'thisisanerror' is not defined");
}
 

@POST
@Path("/reloadHBase/{mediaSourceId}")
public void reloadHBase(@PathParam("mediaSourceId") String mediaSourceId) throws Throwable {
    logger.info("Receive a request for reload hbase-media-source,with id " + mediaSourceId);

    MediaSourceInfo mediaSourceInfo = DataLinkFactory.getObject(MediaSourceService.class).getById(Long.valueOf(mediaSourceId));
    EventBus eventBus = EventBusFactory.getEventBus();
    HBaseConfigClearEvent event = new HBaseConfigClearEvent(new FutureCallback(), mediaSourceInfo);
    eventBus.post(event);
    event.getCallback().get();
    //清空相关Task的mapping缓存
    clearTaskMediaMappingCache(Long.valueOf(mediaSourceId));
}
 

protected AnalysisResult update(List<String> targets,
    boolean keepGoing,
    int loadingPhaseThreads,
    boolean doAnalysis,
    EventBus eventBus) throws Exception {
  return update(
      targets, ImmutableList.<String>of(), keepGoing, loadingPhaseThreads, doAnalysis, eventBus);
}
 

/**
 * Verify that a java_lite_proto_library exposes Starlark providers for the Java code it
 * generates.
 */
@Test
public void testJavaProtosExposeStarlarkProviders() throws Exception {
  scratch.file(
      "proto/extensions.bzl",
      "def _impl(ctx):",
      "  print (ctx.attr.dep[JavaInfo])",
      "custom_rule = rule(",
      "  implementation=_impl,",
      "  attrs={",
      "    'dep': attr.label()",
      "  },",
      ")");
  scratch.file(
      "proto/BUILD",
      "load('//proto:extensions.bzl', 'custom_rule')",
      "load('//tools/build_rules/java_lite_proto_library:java_lite_proto_library.bzl',",
      "      'java_lite_proto_library')",
      "proto_library(",
      "    name = 'proto',",
      "    srcs = [ 'file.proto' ],",
      ")",
      "java_lite_proto_library(name = 'lite_pb2', deps = [':proto'])",
      "custom_rule(name = 'custom', dep = ':lite_pb2')");
  update(
      ImmutableList.of("//proto:custom"),
      /* keepGoing= */ false,
      /* loadingPhaseThreads= */ 1,
      /* doAnalysis= */ true,
      new EventBus());
  // Implicitly check that `update()` above didn't throw an exception. This implicitly checks that
  // ctx.attr.dep.java.{transitive_deps, outputs}, above, is defined.
}
 

@Before
public void setUp() throws Exception {
    data = SystemData.builder()
                     .dataMap(Maps.newHashMap())
                     .errors(1)
                     .processed(100)
                     .build();
    bus = new EventBus();
    bus.register(this);
    jobs = new JobsBeingExecuted(
                                 bus, 10, JobName.Types.SIMPLE);
    pjp = Mockito.mock(ProceedingJoinPoint.class);
}
 

@POST
@Path("/reloadEs/{mediaSourceId}")
public void reloadEs(@PathParam("mediaSourceId") String mediaSourceId) throws Throwable {
    logger.info("Receive a request for reload es-media-source,with id " + mediaSourceId);

    MediaSourceInfo mediaSourceInfo = DataLinkFactory.getObject(MediaSourceService.class).getById(Long.valueOf(mediaSourceId));
    EventBus eventBus = EventBusFactory.getEventBus();
    EsConfigClearEvent event = new EsConfigClearEvent(new FutureCallback(), mediaSourceInfo);
    eventBus.post(event);
    event.getCallback().get();
    //清空相关Task的mapping缓存
    clearTaskMediaMappingCache(Long.valueOf(mediaSourceId));
}
 

public static <T> GenericEvent<T> trigger(String name, EventBus bus, T data, String[] subTypes) {
    GenericEvent<T> event = new GenericEvent<>(GenericEventData.<T>builder()
                                                       .name(name)
                                                       .data(data)
                                                       .subTypes(subTypes)
                                                       .build());
    bus.post(event);
    return event;
}
 

@Test(expected = EntryInSnapshotException.class)
public void testCreateAppendEntriesLogNotEmptyEntryInSnapshot() {
    MemoryLog log = new MemoryLog(
            new MemorySnapshot(3, 2),
            new MemoryEntrySequence(4),
            new EventBus()
    );
    log.appendEntry(1); // 4
    log.createAppendEntriesRpc(
            2, new NodeId("A"), 3, Log.ALL_ENTRIES
    );
}
 

@Test
public void testAppendEntriesFromLeaderSnapshot1() {
    MemoryLog log = new MemoryLog(
            new MemorySnapshot(3, 4),
            new MemoryEntrySequence(4),
            new EventBus()
    );
    Assert.assertTrue(log.appendEntriesFromLeader(3, 4, Collections.emptyList()));
}
 

@Before
public void setup() {
    registry = new MetricRegistry();
    bus = new EventBus();
    config = new Configuration(
                               false, false, false, false, 5, 6, 7, 8, 10, "bob");
    catcher = new MetricsCatcher<>(
                                   registry, bus, config);
}