Add CouchBase sample
This commit is contained in:
David Turanski
@@ -0,0 +1,22 @@
|
||||
== Options
|
||||
|
||||
The **$$couchbase$$** $$sink$$ has the following options:
|
||||
|
||||
|
||||
//tag::configuration-properties[]
|
||||
Properties grouped by prefix:
|
||||
|
||||
|
||||
=== couchbase.consumer
|
||||
|
||||
$$bucket-expression$$:: $$A SpEL expression to specify the bucket.$$ *($$Expression$$, default: `$$<none>$$`)*
|
||||
$$collection-expression$$:: $$A SpEL expression to specify the collection.$$ *($$Expression$$, default: `$$<none>$$`)*
|
||||
$$key-expression$$:: $$A SpEL expression to specify the key.$$ *($$Expression$$, default: `$$<none>$$`)*
|
||||
$$value-expression$$:: $$A SpEL expression to specify the value (default is payload).$$ *($$Expression$$, default: `$$<none>$$`)*
|
||||
|
||||
=== spring.couchbase
|
||||
|
||||
$$connection-string$$:: $$Connection string used to locate the Couchbase cluster.$$ *($$String$$, default: `$$<none>$$`)*
|
||||
$$password$$:: $$Cluster password.$$ *($$String$$, default: `$$<none>$$`)*
|
||||
$$username$$:: $$Cluster username.$$ *($$String$$, default: `$$<none>$$`)*
|
||||
//end::configuration-properties[]
|
||||
@@ -0,0 +1,132 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>io.spring.example</groupId>
|
||||
<artifactId>couchbase-stream-applications</artifactId>
|
||||
<version>0.0.1-SNAPSHOT</version>
|
||||
<relativePath>..</relativePath>
|
||||
</parent>
|
||||
<groupId>io.spring.example</groupId>
|
||||
<artifactId>couchbase-sink</artifactId>
|
||||
<version>0.0.1-SNAPSHOT</version>
|
||||
<name>couchbase-sink</name>
|
||||
<description>Demo Couchbase Sink</description>
|
||||
|
||||
<properties>
|
||||
<test.group>!integration</test.group>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>io.spring.example</groupId>
|
||||
<artifactId>couchbase-consumer</artifactId>
|
||||
<version>0.0.1-SNAPSHOT</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-actuator</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream</artifactId>
|
||||
<type>test-jar</type>
|
||||
<classifier>test-binder</classifier>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>couchbase</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>kafka</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>junit-jupiter</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.stream.app</groupId>
|
||||
<artifactId>stream-applications-test-support</artifactId>
|
||||
<version>3.0.0-RC1</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-failsafe-plugin</artifactId>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.junit.jupiter</groupId>
|
||||
<artifactId>junit-jupiter-engine</artifactId>
|
||||
<version>5.7.0</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<configuration>
|
||||
<groups>integration</groups>
|
||||
<includes>
|
||||
<include>*IntegrationTests.java</include>
|
||||
</includes>
|
||||
<failIfNoTests>true</failIfNoTests>
|
||||
</configuration>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||
<executions>
|
||||
<execution>
|
||||
<id>build-imagage</id>
|
||||
<goals>
|
||||
<goal>build-image</goal>
|
||||
</goals>
|
||||
<phase>pre-integration-test</phase>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-dataflow-apps-docs-plugin</artifactId>
|
||||
<version>1.1.0-SNAPSHOT</version>
|
||||
<executions>
|
||||
<execution>
|
||||
<id>generate-documentation</id>
|
||||
<phase>verify</phase>
|
||||
<goals>
|
||||
<goal>generate-documentation</goal>
|
||||
</goals>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-surefire-plugin</artifactId>
|
||||
<configuration>
|
||||
<groups>!integration</groups>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
@@ -0,0 +1,31 @@
|
||||
/*
|
||||
* Copyright 2020-2020 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package io.spring.example.couchbase.sink;
|
||||
|
||||
import io.spring.example.couchbase.consumer.CouchbaseConsumerConfiguration;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.context.annotation.Import;
|
||||
|
||||
@SpringBootApplication
|
||||
@Import(CouchbaseConsumerConfiguration.class)
|
||||
public class CouchbaseSinkApplication {
|
||||
public static void main(String... args) {
|
||||
new SpringApplication(CouchbaseSinkApplication.class).run(args);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
configuration-properties.classes=io.spring.example.couchbase.consumer.CouchbaseConsumerProperties, \
|
||||
org.springframework.boot.autoconfigure.couchbase.CouchbaseProperties
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
management.endpoints.web.exposure.include=health,info,bindings
|
||||
spring.cloud.function.definition=couchbaseConsumer
|
||||
spring.cloud.stream.function.bindings.couchbaseConsumer-in-0=input
|
||||
@@ -0,0 +1,79 @@
|
||||
/*
|
||||
* Copyright 2020-2020 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package io.spring.example.couchbase.sink;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import com.couchbase.client.java.Cluster;
|
||||
import io.spring.example.couchbase.sink.domain.User;
|
||||
import org.junit.jupiter.api.BeforeAll;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.testcontainers.couchbase.BucketDefinition;
|
||||
import org.testcontainers.couchbase.CouchbaseContainer;
|
||||
import org.testcontainers.junit.jupiter.Container;
|
||||
import org.testcontainers.junit.jupiter.Testcontainers;
|
||||
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
import org.springframework.cloud.stream.binder.test.InputDestination;
|
||||
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.awaitility.Awaitility.await;
|
||||
|
||||
@Testcontainers
|
||||
public class CouchbaseSinkApplicationTests {
|
||||
@Container
|
||||
static CouchbaseContainer container = new CouchbaseContainer("couchbase/server:6.6.0")
|
||||
.withBucket(new BucketDefinition("test"));
|
||||
|
||||
static Map<String, Object> connectProperties = new HashMap<>();
|
||||
|
||||
@BeforeAll
|
||||
static void initialize() {
|
||||
connectProperties.put("spring.couchbase.connection-string", container.getConnectionString());
|
||||
connectProperties.put("spring.couchbase.username", container.getUsername());
|
||||
connectProperties.put("spring.couchbase.password", container.getPassword());
|
||||
}
|
||||
|
||||
@Test
|
||||
void test() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration
|
||||
.getCompleteConfiguration(CouchbaseSinkApplication.class, CouchbaseSinkApplication.class))
|
||||
.web(WebApplicationType.NONE)
|
||||
.properties(connectProperties)
|
||||
.run("--couchbase.consumer.bucketExpression='test'",
|
||||
"--couchbase.consumer.keyExpression=payload.email")) {
|
||||
InputDestination inputDestination = context.getBean(InputDestination.class);
|
||||
Cluster cluster = context.getBean(Cluster.class);
|
||||
inputDestination.send(new GenericMessage<>(new User("Bart Simpson", "bart@simpsons.com")));
|
||||
|
||||
await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> {
|
||||
User user = cluster.bucket("test").defaultCollection().get("bart@simpsons.com")
|
||||
.contentAs(User.class);
|
||||
assertThat(user).isNotNull();
|
||||
assertThat(user.getName()).isEqualTo("Bart Simpson");
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
/*
|
||||
* Copyright 2020-2020 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package io.spring.example.couchbase.sink.domain;
|
||||
|
||||
import java.util.Objects;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
|
||||
|
||||
@JsonIgnoreProperties(ignoreUnknown = true)
|
||||
public class User {
|
||||
private String name;
|
||||
|
||||
private String email;
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public void setName(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
public String getEmail() {
|
||||
return email;
|
||||
}
|
||||
|
||||
public void setEmail(String email) {
|
||||
this.email = email;
|
||||
}
|
||||
|
||||
public User() {
|
||||
}
|
||||
|
||||
public User(String name, String email) {
|
||||
this.name = name;
|
||||
this.email = email;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (this == o) {
|
||||
return true;
|
||||
}
|
||||
if (o == null || getClass() != o.getClass()) {
|
||||
return false;
|
||||
}
|
||||
User user = (User) o;
|
||||
return Objects.equals(name, user.name) &&
|
||||
Objects.equals(email, user.email);
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(name, email);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,108 @@
|
||||
/*
|
||||
* Copyright 2020-2020 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package io.spring.example.couchbase.sink.integration;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
import com.couchbase.client.java.Cluster;
|
||||
import com.couchbase.client.java.ClusterOptions;
|
||||
import com.couchbase.client.java.kv.ExistsResult;
|
||||
import io.spring.example.couchbase.sink.domain.User;
|
||||
import org.junit.jupiter.api.AfterAll;
|
||||
import org.junit.jupiter.api.BeforeAll;
|
||||
import org.junit.jupiter.api.Tag;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.testcontainers.containers.wait.strategy.Wait;
|
||||
import org.testcontainers.couchbase.BucketDefinition;
|
||||
import org.testcontainers.couchbase.CouchbaseContainer;
|
||||
import org.testcontainers.couchbase.CouchbaseService;
|
||||
import org.testcontainers.junit.jupiter.Container;
|
||||
import org.testcontainers.shaded.com.fasterxml.jackson.core.JsonProcessingException;
|
||||
import org.testcontainers.shaded.com.fasterxml.jackson.databind.ObjectMapper;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.cloud.stream.app.test.integration.StreamAppContainer;
|
||||
import org.springframework.cloud.stream.app.test.integration.TestTopicSender;
|
||||
import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaStreamAppTest;
|
||||
import org.springframework.cloud.stream.app.test.integration.kafka.KafkaConfig;
|
||||
import org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamAppContainer;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.awaitility.Awaitility.await;
|
||||
import static org.springframework.cloud.stream.app.test.integration.AppLog.appLog;
|
||||
|
||||
@KafkaStreamAppTest
|
||||
@Tag("integration")
|
||||
public class CouchbaseSinkIntegrationTests {
|
||||
|
||||
static StreamAppContainer sink = new KafkaStreamAppContainer("couchbase-sink:0.0.1-SNAPSHOT");
|
||||
|
||||
@Container
|
||||
static CouchbaseContainer container = new CouchbaseContainer("couchbase/server:6.6.0")
|
||||
.withEnabledServices(CouchbaseService.INDEX, CouchbaseService.KV, CouchbaseService.QUERY,
|
||||
CouchbaseService.SEARCH)
|
||||
.withNetwork(KafkaConfig.kafka.getNetwork())
|
||||
.withNetworkAliases("couchbase-server")
|
||||
.withBucket(new BucketDefinition("test"));
|
||||
|
||||
static Cluster cluster;
|
||||
|
||||
@Autowired
|
||||
TestTopicSender testTopicSender;
|
||||
|
||||
@BeforeAll
|
||||
static void initialize() {
|
||||
await().until(() -> container.isRunning());
|
||||
String connectionString = "couchbase://couchbase-server";
|
||||
sink.waitingFor(Wait.forLogMessage(".*Started CouchbaseSink.*", 1))
|
||||
.withLogConsumer(appLog("couchbase-sink"))
|
||||
.withCommand(
|
||||
"--spring.couchbase.connection-string=couchbase://couchbase-server",
|
||||
"--spring.couchbase.username=" + container.getUsername(),
|
||||
"--spring.couchbase.password=" + container.getPassword(),
|
||||
"--couchbase.consumer.bucket-expression='test'",
|
||||
"--couchbase.consumer.key-expression=payload.email")
|
||||
.start();
|
||||
|
||||
cluster = Cluster.connect(container.getConnectionString(),
|
||||
ClusterOptions.clusterOptions(container.getUsername(), container.getPassword()));
|
||||
}
|
||||
|
||||
@AfterAll
|
||||
static void stop() {
|
||||
sink.stop();
|
||||
}
|
||||
|
||||
@Test
|
||||
void test() throws JsonProcessingException {
|
||||
ObjectMapper objectMapper = new ObjectMapper();
|
||||
testTopicSender.send(sink.getInputDestination(),
|
||||
objectMapper.writeValueAsString(new User("Bart Simpson", "bart@simpsons.com")));
|
||||
|
||||
await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> {
|
||||
ExistsResult result = cluster.bucket("test").defaultCollection().exists("bart@simpsons.com");
|
||||
assertThat(result.exists()).isTrue();
|
||||
});
|
||||
|
||||
User user = objectMapper.readValue(
|
||||
cluster.bucket("test").defaultCollection().get("bart@simpsons.com").contentAs(String.class),
|
||||
User.class);
|
||||
|
||||
assertThat(user.getName()).isEqualTo("Bart Simpson");
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
<configuration>
|
||||
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
|
||||
<encoder>
|
||||
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n</pattern>
|
||||
</encoder>
|
||||
</appender>
|
||||
|
||||
<root level="info">
|
||||
<appender-ref ref="STDOUT"/>
|
||||
</root>
|
||||
|
||||
<logger name="org.testcontainers" level="INFO"/>
|
||||
<logger name="com.github.dockerjava" level="WARN"/>
|
||||
</configuration>
|
||||
Reference in New Issue
Block a user