Java 类akka.actor.ExtendedActorSystem 实例源码

@Test
public void testDoLoadAsyncWithAkkaSerializedSnapshot() throws IOException {
    SnapshotSerializer snapshotSerializer = new SnapshotSerializer((ExtendedActorSystem) system);

    String name = toSnapshotName(PERSISTENCE_ID, 1, 1000);
    try (FileOutputStream fos = new FileOutputStream(new File(SNAPSHOT_DIR, name))) {
        fos.write(snapshotSerializer.toBinary(new Snapshot("one")));
    }

    SnapshotMetadata metadata = new SnapshotMetadata(PERSISTENCE_ID, 1, 1000);

    JavaTestKit probe = new JavaTestKit(system);
    snapshotStore.tell(new LoadSnapshot(PERSISTENCE_ID,
            SnapshotSelectionCriteria.latest(), Long.MAX_VALUE), probe.getRef());
    LoadSnapshotResult result = probe.expectMsgClass(LoadSnapshotResult.class);
    Option<SelectedSnapshot> possibleSnapshot = result.snapshot();

    assertEquals("SelectedSnapshot present", TRUE, possibleSnapshot.nonEmpty());
    assertEquals("SelectedSnapshot metadata", metadata, possibleSnapshot.get().metadata());
    assertEquals("SelectedSnapshot snapshot", "one", possibleSnapshot.get().snapshot());
}

@Override
public Ignite apply(ExtendedActorSystem system) {
    final IgniteProperties properties = igniteConfigProvider.apply(system);
    final IgniteConfiguration igniteConfiguration = new IgniteConfiguration();
    igniteConfiguration.setClientMode(properties.isClientNode());
    // durable file memory persistence
    if (properties.isEnableFilePersistence()) {
        DataStorageConfiguration persistentStoreConfiguration = new DataStorageConfiguration();
        persistentStoreConfiguration.setStoragePath(properties.getIgnitePersistenceFilePath() + "/data/store");
        persistentStoreConfiguration.setWalArchivePath(properties.getIgnitePersistenceFilePath() + "./data/walArchive");
        igniteConfiguration.setDataStorageConfiguration(persistentStoreConfiguration);
    }
    // connector configuration
    final ConnectorConfiguration connectorConfiguration = new ConnectorConfiguration();
    connectorConfiguration.setPort(properties.getIgniteConnectorPort());
    // common ignite configuration
    igniteConfiguration.setMetricsLogFrequency(properties.getMetricsLogFrequency());
    igniteConfiguration.setQueryThreadPoolSize(properties.getQueryThreadPoolSize());
    igniteConfiguration.setDataStreamerThreadPoolSize(properties.getDataStreamerThreadPoolSize());
    igniteConfiguration.setManagementThreadPoolSize(properties.getManagementThreadPoolSize());
    igniteConfiguration.setPublicThreadPoolSize(properties.getPublicThreadPoolSize());
    igniteConfiguration.setSystemThreadPoolSize(properties.getSystemThreadPoolSize());
    igniteConfiguration.setRebalanceThreadPoolSize(properties.getRebalanceThreadPoolSize());
    igniteConfiguration.setAsyncCallbackPoolSize(properties.getAsyncCallbackPoolSize());
    igniteConfiguration.setPeerClassLoadingEnabled(properties.isPeerClassLoadingEnabled());

    final BinaryConfiguration binaryConfiguration = new BinaryConfiguration();
    binaryConfiguration.setCompactFooter(false);
    igniteConfiguration.setBinaryConfiguration(binaryConfiguration);
    // cluster tcp configuration
    final TcpDiscoverySpi tcpDiscoverySpi = new TcpDiscoverySpi();
    final TcpDiscoveryVmIpFinder tcpDiscoveryVmIpFinder = new TcpDiscoveryVmIpFinder();
    // need to be changed when it come to real cluster configuration
    tcpDiscoveryVmIpFinder.setAddresses(Arrays.asList(properties.getTcpDiscoveryAddresses() + properties.getIgniteServerPortRange()));
    tcpDiscoverySpi.setIpFinder(tcpDiscoveryVmIpFinder);
    igniteConfiguration.setDiscoverySpi(new TcpDiscoverySpi());
    final Ignite ignite = Ignition.start(igniteConfiguration);
    Runtime.getRuntime().addShutdownHook(new Thread(ignite::close));
    return ignite;
}

private Object tryDeserializeAkkaSnapshot(final File file) throws IOException {
    LOG.debug("tryDeserializeAkkaSnapshot {}", file);

    // The snapshot was probably previously stored via akka's LocalSnapshotStore which wraps the data
    // in a Snapshot instance and uses the SnapshotSerializer to serialize it to a byte[]. So we'll use
    // the SnapshotSerializer to try to de-serialize it.

    SnapshotSerializer snapshotSerializer = new SnapshotSerializer((ExtendedActorSystem) context().system());

    try (InputStream in = new BufferedInputStream(new FileInputStream(file))) {
        return ((Snapshot)snapshotSerializer.fromBinary(ByteStreams.toByteArray(in))).data();
    }
}

@Override
@Before
public void setUp() throws Exception {
    system = ActorSystem.apply();
    JavaSerializer.currentSystem().value_$eq((ExtendedActorSystem) system);
    super.setUp();
}

public DbmsLayerProvider createExtension(ExtendedActorSystem extendedActorSystem) {
    DbmsLayerProvider provider = null;
    try {
        provider = new DbmsLayerProvider(extendedActorSystem);
    } catch (Exception exc) {
        exc.printStackTrace();
        System.err.println("Unable to init DbmsLayer, halting: " + exc.getMessage());
        System.exit(123);
    }
    return provider;
}

private void handleRequestAssemblerMessage(final Object message) {
    dispatchers.getDispatcher(DispatcherType.Serialization).execute(() -> {
        JavaSerializer.currentSystem().value_$eq((ExtendedActorSystem) context().system());
        requestMessageAssembler.handleMessage(message, self());
    });
}

@Before
public void setUp() {
    JavaSerializer.currentSystem().value_$eq((ExtendedActorSystem) actorSystem);
}

@Before
public void setUp() {
    JavaSerializer.currentSystem().value_$eq((ExtendedActorSystem) actorSystem);
}

@Before
public void setUp() {
    JavaSerializer.currentSystem().value_$eq((ExtendedActorSystem) SYSTEM);
}

@Before
public void setUp() {
    JavaSerializer.currentSystem().value_$eq((ExtendedActorSystem) SYSTEM);
}

public ApplicationStatusPersistenceQuery(ExtendedActorSystem system) {
    super(system);
}

public SettingsImpl createExtension(ExtendedActorSystem system) {
    return new SettingsImpl(system.settings().config());
}

public Replication createExtension(ExtendedActorSystem system) {
    return new Replication(system);
}

public SharedActorMaterializer createExtension(ExtendedActorSystem system) {
    return new SharedActorMaterializer(system);
}

private SharedActorMaterializer(ExtendedActorSystem system) {
    this.materializer = ActorMaterializer.create(system);
}

/**
 * Is used by Akka to instantiate the Extension identified by this
 * ExtensionId, internal use only.
 */
@Override
public SpringExt createExtension(ExtendedActorSystem system) {
    return new SpringExt();
}

public ApplicationStatusPersistenceQuery(ExtendedActorSystem system) {
    super(system);
}

@Override
public AkkaExtension createExtension(ExtendedActorSystem system) {
    return new AkkaExtension();
}

@Override
public PolyguiceExtension createExtension(ExtendedActorSystem system) {
    return new PolyguiceExtension(compCtxt);
}

/**
 * Is used by Akka to instantiate the Extension identified by this
 * ExtensionId, internal use only.
 */
@Override
public SpringExt createExtension(ExtendedActorSystem system) {
  return new SpringExt();
}

public GuiceExt createExtension(ExtendedActorSystem system) {
    return new GuiceExt();
}

public SpringExt createExtension(ExtendedActorSystem system) {
    return new SpringExt();
}

/**
 * Is used by Akka to instantiate the Extension identified by this
 * ExtensionId, internal use only.
 */
@Override
public SpringExt createExtension(ExtendedActorSystem system) {
    return new SpringExt();
}

public MySQLJDBCSettingsImpl createExtension(ExtendedActorSystem system) {
    return new MySQLJDBCSettingsImpl(system.settings().config());
}

/**
 * Is used by Akka to instantiate the Extension identified by this
 * ExtensionId, internal use only.
   * @param system
   * @return
 */
@Override
public SpringExt createExtension(ExtendedActorSystem system) {
  return new SpringExt();
}