#528 - Add Redis Streams example.

closes: #528
Original pull request: #529.

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<String, MapRecord<String, String, String>> streamReceiver() {
+		return StreamReceiver.create(factory);
+	}
+
+	/**
+	 * Clear database before shut down.
+	 */
+	public @PreDestroy
+	void flushTestDb() {
+		factory.getReactiveConnection().serverCommands().flushDb().then().as(StepVerifier::create).verifyComplete();
+	}
+}

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<String, MapRecord<String, String, String>> streamReceiver;
+
+	ReactiveStreamOperations<String, String, String> 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<MapRecord<String, String, String>> 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));
+	}
+}

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<String, MapRecord<String, String, String>> streamMessageListenerContainer() {
+		return StreamMessageListenerContainer.create(factory);
+	}
+
+	/**
+	 * Clear database before shut down.
+	 */
+	public @PreDestroy
+	void flushTestDb() {
+		factory.getConnection().flushDb();
+	}
+}

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<String, MapRecord<String, String, String>> messageListenerContainer;
+
+	StreamOperations<String, String, String> 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<MapRecord<String, String, String>> fromStart = streamOps.read(fromStart(SensorData.KEY));
+		assertThat(fromStart).hasSize(3).extracting(MapRecord::getId).containsExactly(fixedId1, fixedId2, autogeneratedId);
+
+		// XREAD resume after
+		List<MapRecord<String, String, String>> 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);
+	}
+}

redis/streams/src/test/resources/application.properties

@@ -0,0 +1 @@
+logging.level.root=WARN