diff --git a/README.md b/README.md index a9ea2d51..2e83cee8 100644 --- a/README.md +++ b/README.md @@ -51,6 +51,7 @@ We have separate folders for the samples of individual modules: * `reactive` - Example project to show reactive template support. * `repositories` - Example demonstrating Spring Data repository abstraction on top of Redis. * `sentinel` - Example for Redis Sentinel support. +* `streams` - Example for [Redis Streams](https://redis.io/topics/streams-intro) support. ## Spring Data for Apache Solr diff --git a/redis/pom.xml b/redis/pom.xml index c28b4c7e..96f9dfe4 100644 --- a/redis/pom.xml +++ b/redis/pom.xml @@ -21,6 +21,7 @@ reactive repositories sentinel + streams util diff --git a/redis/streams/README.md b/redis/streams/README.md new file mode 100644 index 00000000..9720f59b --- /dev/null +++ b/redis/streams/README.md @@ -0,0 +1,61 @@ +# Spring Data Redis - Streams Examples + +The [Redis Stream](https://redis.io/topics/streams-intro) is a new data type introduced with Redis 5.0 modelling log data structure. +Spring Data Redis supports _Redis Streams_ via both the imperative and the reactive API. + +## Imperative API + +**Basic Usage** +```java +@Autowired +RedisTemplate template; + +StringRecord record = StreamRecords.string(…) + .withStreamKey("my-stream"); +RecordId id = template.streamOps().add(record); + +List<...> records = template.streamOps().read(count(2), from(id)); +``` + +**ContinuousRead Read** +```java +@Autowired +RedisConnectionFactory factory; + +StreamListener listener = + (msg) -> { + // ... + }; + +StreamMessageListenerContainer container = StreamMessageListenerContainer.create(factory)); + +container.receive(StreamOffset.fromStart("my-stream"), listener); +``` + +## Reactive API + +**Basic Usage** +```java +@Autowired +ReactiveRedisTemplate template; + +StringRecord record = StreamRecords.string(…) + .withStreamKey("my-stream"); +Mono id = template.streamOps().add(record); + +Flux<...> records = template.streamOps().read(count(2), from(id)); +``` + +**ContinuousRead Read** +```java +@Autowired +ReactiveRedisConnectionFactory factory; + +StreamReceiver receiver = StreamReceiver.create(factory)); + +container.receive(StreamOffset.fromStart("my-stream")) + .doOnNext((msg) -> { + // ... + }) + .subscribe(); +``` diff --git a/redis/streams/pom.xml b/redis/streams/pom.xml new file mode 100644 index 00000000..c8f7e896 --- /dev/null +++ b/redis/streams/pom.xml @@ -0,0 +1,31 @@ + + 4.0.0 + + spring-data-redis-streams-example + Spring Data Redis - Streams Example + + + org.springframework.data.examples + spring-data-redis-examples + 2.0.0.BUILD-SNAPSHOT + ../pom.xml + + + + + + io.projectreactor + reactor-test + test + + + ${project.groupId} + spring-data-redis-example-utils + ${project.version} + test + + + + + diff --git a/redis/streams/src/main/java/example/springdata/redis/SensorData.java b/redis/streams/src/main/java/example/springdata/redis/SensorData.java new file mode 100644 index 00000000..6765f9c3 --- /dev/null +++ b/redis/streams/src/main/java/example/springdata/redis/SensorData.java @@ -0,0 +1,50 @@ +/* + * Copyright 2019 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 example.springdata.redis; + +import java.util.LinkedHashMap; +import java.util.Map; + +import org.springframework.data.redis.connection.stream.RecordId; +import org.springframework.data.redis.connection.stream.StreamRecords; +import org.springframework.data.redis.connection.stream.StringRecord; + +/** + * @author Christoph Strobl + */ +public final class SensorData { + + public static final String KEY = "my-stream"; + + public static final StringRecord RECORD_1234_0 = SensorData.create("S-12", "18", "r2d2").withId(RecordId.of("1234-0")); + public static final StringRecord RECORD_1234_1 = SensorData.create("S-13", "9", "c3o0").withId(RecordId.of("1234-1")); + public static final StringRecord RECORD_1235_0 = SensorData.create("S-13", "18.2", "bb8").withId(RecordId.of("1235-0")); + + public static StringRecord create(String sensor, String temperature, String checksum) { + return StreamRecords.string(sensorData(sensor, temperature, checksum)).withStreamKey(KEY); + } + + private static Map sensorData(String... values) { + + Map data = new LinkedHashMap<>(); + data.put("sensor-id", values[0]); + data.put("temperature", values[1]); + if (values.length >= 3 && values[2] != null) { + data.put("checksum", values[2]); + } + return data; + } +} diff --git a/redis/streams/src/main/java/example/springdata/redis/sync/CapturingStreamListener.java b/redis/streams/src/main/java/example/springdata/redis/sync/CapturingStreamListener.java new file mode 100644 index 00000000..fbd21b25 --- /dev/null +++ b/redis/streams/src/main/java/example/springdata/redis/sync/CapturingStreamListener.java @@ -0,0 +1,67 @@ +/* + * Copyright 2019 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 example.springdata.redis.sync; + +import java.util.concurrent.BlockingDeque; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.atomic.AtomicInteger; + +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.stream.StreamListener; + +/** + * @author Christoph Strobl + */ +public class CapturingStreamListener implements StreamListener> { + + private AtomicInteger counter = new AtomicInteger(0); + private BlockingDeque> deque = new LinkedBlockingDeque<>(); + + private CapturingStreamListener() { + } + + @Override + public void onMessage(MapRecord record) { + + deque.add(record); + counter.incrementAndGet(); + } + + /** + * @return new instance of {@link CapturingStreamListener}. + */ + public static CapturingStreamListener create() { + return new CapturingStreamListener(); + } + + /** + * @return the total number or records captured so far. + */ + public int recordsReceived() { + return counter.get(); + } + + /** + * Retrieves and removes the head of the queue waiting if + * necessary until an element becomes available. + * + * @return + * @throws InterruptedException if interrupted while waiting + */ + public MapRecord take() throws InterruptedException { + return deque.take(); + } +} diff --git a/redis/streams/src/test/java/example/springdata/redis/reactive/ReactiveRedisTestConfiguration.java b/redis/streams/src/test/java/example/springdata/redis/reactive/ReactiveRedisTestConfiguration.java new file mode 100644 index 00000000..3f1b7bd6 --- /dev/null +++ b/redis/streams/src/test/java/example/springdata/redis/reactive/ReactiveRedisTestConfiguration.java @@ -0,0 +1,48 @@ +/* + * Copyright 2019 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 example.springdata.redis.reactive; + +import javax.annotation.PreDestroy; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.Bean; +import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory; +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.stream.StreamReceiver; +import reactor.test.StepVerifier; + +/** + * @author Christoph Strobl + */ +@SpringBootApplication +public class ReactiveRedisTestConfiguration { + + @Autowired ReactiveRedisConnectionFactory factory; + + @Bean + StreamReceiver> streamReceiver() { + return StreamReceiver.create(factory); + } + + /** + * Clear database before shut down. + */ + public @PreDestroy + void flushTestDb() { + factory.getReactiveConnection().serverCommands().flushDb().then().as(StepVerifier::create).verifyComplete(); + } +} diff --git a/redis/streams/src/test/java/example/springdata/redis/reactive/ReactiveStreamApiTests.java b/redis/streams/src/test/java/example/springdata/redis/reactive/ReactiveStreamApiTests.java new file mode 100644 index 00000000..ab3369c9 --- /dev/null +++ b/redis/streams/src/test/java/example/springdata/redis/reactive/ReactiveStreamApiTests.java @@ -0,0 +1,137 @@ +/* + * Copyright 2019 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 example.springdata.redis.reactive; + +import static org.assertj.core.api.Assertions.*; +import static org.springframework.data.redis.connection.stream.StreamOffset.*; + +import java.time.Duration; + +import example.springdata.redis.SensorData; +import example.springdata.redis.test.util.RequiresRedisServer; +import org.junit.Before; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.data.redis.RedisSystemException; +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.ReadOffset; +import org.springframework.data.redis.connection.stream.RecordId; +import org.springframework.data.redis.connection.stream.StreamOffset; +import org.springframework.data.redis.core.ReactiveStreamOperations; +import org.springframework.data.redis.core.ReactiveStringRedisTemplate; +import org.springframework.data.redis.stream.StreamReceiver; +import org.springframework.test.context.junit4.SpringRunner; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +/** + * @author Christoph Strobl + */ +@RunWith(SpringRunner.class) +@SpringBootTest +public class ReactiveStreamApiTests { + + public static @ClassRule RequiresRedisServer server = RequiresRedisServer.onLocalhost().atLeast("5.0"); + + @Autowired ReactiveStringRedisTemplate template; + @Autowired StreamReceiver> streamReceiver; + + ReactiveStreamOperations streamOps; + + @Before + public void setUp() { + + // clear all + template.getConnectionFactory().getReactiveConnection() + .serverCommands().flushAll() + .then().as(StepVerifier::create) + .verifyComplete(); + + streamOps = template.opsForStream(); + } + + @Test + public void basics() { + + // XADD with fixed id + streamOps.add(SensorData.RECORD_1234_0) + .as(StepVerifier::create) + .expectNext(SensorData.RECORD_1234_0.getId()).verifyComplete(); + + streamOps.add(SensorData.RECORD_1234_1) + .as(StepVerifier::create) + .expectNext(SensorData.RECORD_1234_1.getId()).verifyComplete(); + + // XLEN + streamOps.size(SensorData.KEY) + .as(StepVerifier::create) + .expectNext(2L).verifyComplete(); + + // XADD errors when timestamp is less then last inserted + streamOps.add(SensorData.create("1234", "19.8", "invalid").withId(RecordId.of("0-0"))) + .as(StepVerifier::create) + .verifyError(RedisSystemException.class); + + // XADD with autogenerated id + streamOps.add(SensorData.create("1234", "19.8", null)) + .as(StepVerifier::create) + .consumeNextWith(autogeneratedId -> autogeneratedId.getValue().endsWith("-0")).verifyComplete(); + + streamOps.size(SensorData.KEY) + .as(StepVerifier::create) + .expectNext(3L).verifyComplete(); + + // XREAD from start + streamOps.read(fromStart(SensorData.KEY)) + .map(MapRecord::getId) + .as(StepVerifier::create) + .expectNext(SensorData.RECORD_1234_0.getId(), SensorData.RECORD_1234_1.getId()) + .expectNextCount(1).verifyComplete(); + + // XREAD resume after + streamOps.read(StreamOffset.create(SensorData.KEY, ReadOffset.from(SensorData.RECORD_1234_1.getId()))) + .as(StepVerifier::create) + .expectNextCount(1).verifyComplete(); + } + + @Test + public void continuousRead() { + + Flux> messages = streamReceiver.receive(fromStart(SensorData.KEY)); + + messages.as(StepVerifier::create) + .then(() -> + streamOps.add(SensorData.RECORD_1234_0) + .then(streamOps.add(SensorData.RECORD_1234_1)) + .subscribe()) + .consumeNextWith(it -> { + assertThat(it.getId()).isEqualTo(SensorData.RECORD_1234_0.getId()); + }) + .consumeNextWith(it -> { + assertThat(it.getId()).isEqualTo(SensorData.RECORD_1234_1.getId()); + }) + .then(() -> streamOps.add(SensorData.RECORD_1235_0) + .subscribe()) + .consumeNextWith(it -> { + assertThat(it.getId()).isEqualTo(SensorData.RECORD_1235_0.getId()); + }) + .thenCancel() + .verify(Duration.ofSeconds(5)); + } +} diff --git a/redis/streams/src/test/java/example/springdata/redis/sync/RedisTestConfiguration.java b/redis/streams/src/test/java/example/springdata/redis/sync/RedisTestConfiguration.java new file mode 100644 index 00000000..59d28599 --- /dev/null +++ b/redis/streams/src/test/java/example/springdata/redis/sync/RedisTestConfiguration.java @@ -0,0 +1,47 @@ +/* + * Copyright 2019 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 example.springdata.redis.sync; + +import javax.annotation.PreDestroy; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.Bean; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.stream.StreamMessageListenerContainer; + +/** + * @author Christoph Strobl + */ +@SpringBootApplication +public class RedisTestConfiguration { + + @Autowired RedisConnectionFactory factory; + + @Bean + StreamMessageListenerContainer> streamMessageListenerContainer() { + return StreamMessageListenerContainer.create(factory); + } + + /** + * Clear database before shut down. + */ + public @PreDestroy + void flushTestDb() { + factory.getConnection().flushDb(); + } +} diff --git a/redis/streams/src/test/java/example/springdata/redis/sync/SyncStreamApiTests.java b/redis/streams/src/test/java/example/springdata/redis/sync/SyncStreamApiTests.java new file mode 100644 index 00000000..4219bfcd --- /dev/null +++ b/redis/streams/src/test/java/example/springdata/redis/sync/SyncStreamApiTests.java @@ -0,0 +1,127 @@ +/* + * Copyright 2019 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 example.springdata.redis.sync; + +import static org.assertj.core.api.Assertions.*; +import static org.springframework.data.redis.connection.stream.StreamOffset.*; + +import java.util.List; +import java.util.concurrent.TimeUnit; + +import example.springdata.redis.SensorData; +import example.springdata.redis.test.util.RequiresRedisServer; +import org.junit.Before; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.data.redis.RedisSystemException; +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.ReadOffset; +import org.springframework.data.redis.connection.stream.RecordId; +import org.springframework.data.redis.connection.stream.StreamOffset; +import org.springframework.data.redis.core.StreamOperations; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.stream.StreamMessageListenerContainer; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * @author Christoph Strobl + */ +@RunWith(SpringRunner.class) +@SpringBootTest +public class SyncStreamApiTests { + + public static @ClassRule RequiresRedisServer server = RequiresRedisServer.onLocalhost().atLeast("5.0"); + + @Autowired StringRedisTemplate template; + @Autowired StreamMessageListenerContainer> messageListenerContainer; + + StreamOperations streamOps; + + @Before + public void setUp() { + + // clear all + template.getConnectionFactory().getConnection().flushAll(); + + streamOps = template.opsForStream(); + } + + @Test + public void basics() { + + // XADD with fixed id + RecordId fixedId1 = streamOps.add(SensorData.RECORD_1234_0); + assertThat(fixedId1).isEqualTo(SensorData.RECORD_1234_0.getId()); + + RecordId fixedId2 = streamOps.add(SensorData.RECORD_1234_1); + assertThat(fixedId2).isEqualTo(SensorData.RECORD_1234_1.getId()); + + // XLEN + assertThat(streamOps.size(SensorData.KEY)).isEqualTo(2L); + + // XADD errors when timestamp is less then last inserted + assertThatExceptionOfType(RedisSystemException.class).isThrownBy(() -> { + streamOps.add(SensorData.create("1234", "19.8", "invalid").withId(RecordId.of("0-0"))); + }).withMessageContaining("equal or smaller"); + + // XADD with autogenerated id + RecordId autogeneratedId = streamOps.add(SensorData.create("1234", "19.8", null)); + + assertThat(autogeneratedId.getValue()).endsWith("-0"); + assertThat(streamOps.size(SensorData.KEY)).isEqualTo(3L); + + // XREAD from start + List> fromStart = streamOps.read(fromStart(SensorData.KEY)); + assertThat(fromStart).hasSize(3).extracting(MapRecord::getId).containsExactly(fixedId1, fixedId2, autogeneratedId); + + // XREAD resume after + List> fromOffset = streamOps.read(StreamOffset.create(SensorData.KEY, ReadOffset.from(fixedId2))); + assertThat(fromOffset).hasSize(1).extracting(MapRecord::getId).containsExactly(autogeneratedId); + } + + @Test + public void continuousRead() throws InterruptedException { + + // container autostart is disabled by default + if (!messageListenerContainer.isRunning()) { + messageListenerContainer.start(); + } + + CapturingStreamListener streamListener = CapturingStreamListener.create(); + + // XREAD BLOCK + messageListenerContainer.receive(fromStart(SensorData.KEY), streamListener); + + TimeUnit.MILLISECONDS.sleep(100); + + assertThat(streamListener.recordsReceived()).isEqualTo(0); + + streamOps.add(SensorData.RECORD_1234_0); + streamOps.add(SensorData.RECORD_1234_1); + + assertThat(streamListener.take().getId()).isEqualTo(SensorData.RECORD_1234_0.getId()); + assertThat(streamListener.take().getId()).isEqualTo(SensorData.RECORD_1234_1.getId()); + assertThat(streamListener.recordsReceived()).isEqualTo(2); + + streamOps.add(SensorData.RECORD_1235_0); + + assertThat(streamListener.take().getId()).isEqualTo(SensorData.RECORD_1235_0.getId()); + assertThat(streamListener.recordsReceived()).isEqualTo(3); + } +} diff --git a/redis/streams/src/test/resources/application.properties b/redis/streams/src/test/resources/application.properties new file mode 100644 index 00000000..bac6605c --- /dev/null +++ b/redis/streams/src/test/resources/application.properties @@ -0,0 +1 @@ +logging.level.root=WARN