The Couchbase Developer Experience team is dedicated to providing Couchbase users with a rich open-source ecosystem for their clusters. Recently, we have been working to update the Alpakka Couchbase integration with support for the latest features in Couchbase Server.
The updated Couchbase integration is available as part of the 10.0.1 release of Alpakka. In this post, we will review a simple project that uses the “travel-sample” dataset to provide an overview of basic usage of the integration and its components.
The most comprehensive information about the integration and its API, including additional usage examples, can be found in the Alpakka documentation: https://doc.akka.io/libraries/alpakka/current/couchbase.html
Pré-requisitos
- Couchbase cluster with travel-sample bucket
- Akka CLI
- Java 21, Eclipse Adoptium is recommended by Akka developers
- Apache Maven version 3.9 or later
- enrolar command-line tool
- Git or GitHub Desktop
Note: The Akka dependencies are available from Akka’s secure library repository. To access them you need to use a secure, tokenized URL as specified at https://account.akka.io/token.
In order to demonstrate using Couchbase in Alpakka, the project uses the travel-sample dataset that can be added to any Couchbase cluster from its settings. More information about it can be found in Сouchbase documentation: https://docs.couchbase.com/java-sdk/current/ref/travel-app-data-model.html
The sample project implements two endpoints. One of them lets us search for airports by city, country, airport name, or its international codes and the other returns full information about an airport by its identifier.
The source code for our example project can be found at this URL: https://github.com/couchbaselabs/alpakka-example
The following command can be used to clone it with git:
clone do git https://github.com/couchbaselabs/alpakka-example.git
Alpakka Overview
O Alpakka project is an initiative to implement stream-aware and reactive integration pipelines for Java and Scala. It is built on top of Akka Streams, and has been designed from the ground up to understand streaming natively and provide a DSL for reactive and stream-oriented programming, with built-in support for backpressure.
Akka Streams are structured around three main components: Source, Sink, and Flow. A Source emits data, a Sink consumes it, and a Flow transforms the data as it moves through the stream, allowing for efficient and asynchronous processing.
Our integration provides all three types of components:
- CouchbaseSource component performs Couchbase SQL++ queries and uses the results as source for data in a stream
- CouchbaseFlow fetches documents by their identifiers, and also upserts, replaces and deletes them
- CouchbaseSink provides similar CouchbaseFlow sets of operations that can be used as last step in a stream
Alpakka-Couchbase Configuration
The project source code contains file src/main/resources/application.conf with the following couchbase configuration:
|
1 2 3 4 5 6 7 |
alpakka.couchbase { session { nodes = ["localhost"] username = "Administrator" password = "password" } } |
This configuration should work with a single-node developer cluster that uses default settings and runs on the same machine as the Alpakka application. Alternatively, it may be modified with credentials for a Couchbase Capella cloud-hosted cluster. More information about credentials and connecting to Couchbase can be found in Couchbase Capella (https://docs.couchbase.com/cloud/get-started/connect.html) and Couchbase Server (https://docs.couchbase.com/server/current/guides/connect.html#connecting-via-client) documentation.
Airport Controller
The sample code that interacts with a configured Couchbase connection can be found in the AirportController class, located at src/main/java/travelsample/api/AirportController.java. The Alpakka-Couchbase integration uses CouchbaseSessionSettings objects to reference specific Couchbase connections. The session settings object can be reused between different endpoints and actors, so it is created in the constructor of the controller and stored into a field:
|
1 2 3 4 |
public AirportController(ActorSystem system) { sessionSettings = CouchbaseSessionSettings.create(system) .withEnrichAsyncCs(DiscoverySupport.getNodes(system)); } |
O searchAirport method in this class implements the /search?query=<text> endpoint and demonstrates using SQL++ with positional arguments in Alpakka:
|
1 2 3 4 5 6 7 |
String searchTerm = requestContext().queryParams().getString("query").orElseThrow(); return CouchbaseSource.fromQuery( sessionSettings, "travel-sample", "SELECT * FROM inventory.airport where airportname LIKE '%' $1 '%' OR faa = $1 or icao = $1 or city LIKE '%' $1 '%'", QueryOptions.queryOptions().parameters(JsonArray.from(searchTerm) )).runWith(Sink.seq(), system).toCompletableFuture().get(TIMEOUT, TimeUnit.SECONDS); |
To perform a SQL++ query, we first create an instance of CouchbaseSource class and configure it with the information required to connect to the cluster (sessionSettings), the query we want to execute, and arguments for that query passed as a QueryOptions object. Both positional and named arguments are supported according to the Couchbase Java SDK documentation for the Opções de consulta classe.
Then, we use Alpakka’s runSink method on the created source to run the query and pull its results as a stream of objects.
The second controller method, getById, implements the /byId/{id} endpoint, which takes an airport identifier and returns information about an airport stored in a document with that identifier.
|
1 2 3 4 5 6 7 8 9 10 11 |
return Source.from(Collections.singletonList(id)) .via(CouchbaseFlow.fromId( sessionSettings, "travel-sample", "inventory", "Airport" )) .runWith(Sink.seq(), system) .thenApply(List::getFirst) .thenApply(CouchbaseDocument::getDocument) .toCompletableFuture().get(TIMEOUT, TimeUnit.SECONDS); |
We can use Alpakka’s Source::from method with a singleton list containing the requested identifier to create a source instance.
Then we use the Source.via method to route the source Akka stream (in this case of a single identifier) through a CouchbaseFlow object that is configured similarly to the CouchbaseSource class in the searchAirport method. The instance of CouchbaseFlow will accept the requested identifier from the stream, fetch the corresponding document from the configured collection, and return it to the stream, replacing the identifier with an instance of CouchbaseDocument object, which contains both the identifier and associated document.
Combinação CouchbaseSink, CouchbaseFlow, and other documented Couchbase objects as steps in the Akka streams together with actors from other Alpakka components allows for the creation of pipelines that can use Couchbase to store and fetch information, as well as use its powerful features for analytics and search.
Documentation and issue tracking
The Alpakka documentation (https://doc.akka.io/libraries/alpakka/current/couchbase.html) contains more information about using this integration, including upserting, modificando e exclusão documents in a Couchbase cluster.
Any suggestions about the integration or issues with it can be shared via the Alpakka GitHub Issues page.
