Apache Cassandra

libraryDependencies += "org.apache.pekko" %% "pekko-connectors-cassandra" % "1.0.2"

copysource
import org.apache.pekko.actor.ActorSystem;
import org.apache.pekko.stream.connectors.cassandra.CassandraSessionSettings;
import org.apache.pekko.stream.connectors.cassandra.javadsl.CassandraSession;
import org.apache.pekko.stream.connectors.cassandra.javadsl.CassandraSessionRegistry;

ActorSystem system = // ???
CassandraSessionSettings sessionSettings = CassandraSessionSettings.create();
CassandraSession cassandraSession =
    CassandraSessionRegistry.get(system).sessionFor(sessionSettings);

CompletionStage<String> version =
    cassandraSession
        .select("SELECT release_version FROM system.local;")
        .map(row -> row.getString("release_version"))
        .runWith(Sink.head(), system);

copysourceimport org.apache.pekko.stream.connectors.cassandra.javadsl.CassandraSource;

CompletionStage<List<Integer>> select =
    CassandraSource.create(cassandraSession, "SELECT id FROM " + idtable + ";")
        .map(r -> r.getInt("id"))
        .runWith(Sink.seq(), system);

CompletionStage<Integer> select =
    CassandraSource.create(
            cassandraSession, "SELECT * FROM " + idtable + " WHERE id = ?;", value)
        .map(r -> r.getInt("id"))
        .runWith(Sink.head(), system);

copysourceimport com.datastax.oss.driver.api.core.cql.SimpleStatement;
import com.datastax.oss.driver.api.core.cql.Statement;

Statement<?> stmt =
    SimpleStatement.newInstance("SELECT * FROM " + idtable + ";").setPageSize(20);

CompletionStage<List<Integer>> select =
    CassandraSource.create(cassandraSession, stmt)
        .map(r -> r.getInt("id"))
        .runWith(Sink.seq(), system);

copysourceimport org.apache.pekko.NotUsed;
import org.apache.pekko.actor.ActorSystem;
import org.apache.pekko.japi.Function2;
import org.apache.pekko.japi.Pair;
import org.apache.pekko.stream.connectors.cassandra.CassandraWriteSettings;
import org.apache.pekko.stream.connectors.cassandra.javadsl.CassandraFlow;
import org.apache.pekko.stream.connectors.testkit.javadsl.LogCapturingJunit4;
import org.apache.pekko.stream.javadsl.SourceWithContext;
import com.datastax.oss.driver.api.core.cql.BoundStatement;
import com.datastax.oss.driver.api.core.cql.PreparedStatement;

List<Person> persons =
    Arrays.asList(
        new Person(12, "John", "London"),
        new Person(43, "Umberto", "Roma"),
        new Person(56, "James", "Chicago"));

Function2<Person, PreparedStatement, BoundStatement> statementBinder =
    (person, preparedStatement) -> preparedStatement.bind(person.id, person.name, person.city);

CompletionStage<List<Person>> written =
    Source.from(persons)
        .via(
            CassandraFlow.create(
                cassandraSession,
                CassandraWriteSettings.defaults(),
                "INSERT INTO " + table + "(id, name, city) VALUES (?, ?, ?)",
                statementBinder))
        .runWith(Sink.seq(), system);

copysourceSourceWithContext<Person, AckHandle, NotUsed> from = // ???;
CompletionStage<Done> written =
    from.via(
            CassandraFlow.withContext(
                cassandraSession,
                CassandraWriteSettings.defaults(),
                "INSERT INTO " + table + "(id, name, city) VALUES (?, ?, ?)",
                (person, preparedStatement) ->
                    preparedStatement.bind(person.id, person.name, person.city)))
        .asSource()
        .mapAsync(1, pair -> pair.second().ack())
        .runWith(Sink.ignore(), system);

copysource# SPDX-License-Identifier: Apache-2.0

pekko.connectors.cassandra {
  # The implementation of `org.apache.pekko.stream.connectors.cassandra.CqlSessionProvider`
  # used for creating the `CqlSession`.
  # It may optionally have a constructor with an `ClassicActorSystemProvider` and `Config` parameters.
  session-provider = "org.apache.pekko.stream.connectors.cassandra.DefaultSessionProvider"

  # Configure Pekko Discovery by setting a service name
  service-discovery {
    name = ""
    lookup-timeout = 1 s
  }

  # The ExecutionContext to use for the session tasks and future composition.
  session-dispatcher = "pekko.actor.default-dispatcher"

  # Full config path to the Datastax Java driver's configuration section.
  # When connecting to more than one Cassandra cluster different session configuration can be
  # defined with this property.
  # See https://docs.datastax.com/en/developer/java-driver/latest/manual/core/configuration/#quick-overview
  # and https://docs.datastax.com/en/developer/java-driver/latest/manual/core/configuration/reference/
  datastax-java-driver-config = "datastax-java-driver"
}

copysourcedatastax-java-driver {
  basic {
    contact-points = [ "127.0.0.1:9042" ]
    load-balancing-policy.local-datacenter = datacenter1
  }
  advanced.reconnect-on-init = true
}

val PekkoVersion = "1.0.3"
libraryDependencies += "org.apache.pekko" %% "pekko-discovery" % PekkoVersion

copysourcepekko {
  discovery.method = config
}
pekko.discovery.config.services = {
  cassandra-service = {
    endpoints = [
      {
        host = "127.0.0.1"
        port = 9042
      }
    ]
  }
}

// inherit defaults from `pekko.connectors.cassandra` settings
example-with-pekko-discovery: ${pekko.connectors.cassandra} {
  service-discovery.name = "cassandra-service"
}

copysourceCassandraSessionSettings sessionSettings =
    CassandraSessionSettings.create("example-with-pekko-discovery");
CassandraSession session = CassandraSessionRegistry.get(system).sessionFor(sessionSettings);