Repository: redis/lettuce (java)

Issue:
Provide a convenience method to support reading of last message from stream
## Feature Request

<!-- First of all: Have you checked the docs https://github.com/lettuce-io/lettuce-core/wiki, GitHub issues, or Stack Overflow whether someone else has already reported your issue? Maybe the feature already exists?-->
A new way to read from the last element of the stream was introduced to `XREAD` with https://github.com/redis/redis/issues/7388

We might want to add a convenience method to the `XReadArgs.StreamOffset` class.

#### Is your feature request related to a problem? Please describe

The feature could still be used if the `+` sign is used as offset in the `XReadArgs.StreamOffset` constructor. 

Having a dedicated method would make it easier for the consumers of the driver.

#### Describe the solution you'd like

Add a new public static method that creates a `StreamOffset` with the hardcoded offset sign.

#### Describe alternatives you've considered

None.

#### Teachability, Documentation, Adoption, Migration Strategy

Similar to how the existing `.latest(K name)` and `.lastConsumed(K name) `methods were used.


Pull request description:
XREAD support for reading last message from stream
Closes #2861 

Added a convenience utility method to read the last message from a stream

<!--
Thank you for proposing a pull request. This template will guide you through the essential steps necessary for a pull request.
-->
Make sure that:

- [x] You have read the [contribution guidelines](https://github.com/lettuce-io/lettuce-core/blob/main/.github/CONTRIBUTING.md).
- [x] You have created a feature request first to discuss your contribution intent. Please reference the feature request ticket number in the pull request.
- [x] You use the code formatters provided [here](https://github.com/lettuce-io/lettuce-core/blob/main/formatting.xml) and have them applied to your changes. Don’t submit any formatting related changes.
- [x] You submit test cases (unit or integration tests) that back your changes.

 <!--
Great! Live long and prosper.
-->


Patch:
diff --git a/src/main/java/io/lettuce/core/XReadArgs.java b/src/main/java/io/lettuce/core/XReadArgs.java
index 13d0b4938..c08338e2d 100644
--- a/src/main/java/io/lettuce/core/XReadArgs.java
+++ b/src/main/java/io/lettuce/core/XReadArgs.java
@@ -1,11 +1,11 @@
 package io.lettuce.core;
 
-import java.time.Duration;
-
 import io.lettuce.core.internal.LettuceAssert;
 import io.lettuce.core.protocol.CommandArgs;
 import io.lettuce.core.protocol.CommandKeyword;
 
+import java.time.Duration;
+
 /**
  * Argument list builder for the Redis <a href="https://redis.io/commands/xread">XREAD</a> and {@literal XREADGROUP} commands.
  * Static import the methods from {@link XReadArgs.Builder} and call the methods: {@code block(…)} .
@@ -171,7 +171,7 @@ public class XReadArgs implements CompositeArgument {
         }
 
         /**
-         * Read all new arriving elements from the stream identified by {@code name}.
+         * Read all new arriving elements from the stream identified by {@code name} excluding any elements before this call
          *
          * @param name must not be {@code null}.
          * @return the {@link StreamOffset} object without a specific offset.
@@ -183,6 +183,21 @@ public class XReadArgs implements CompositeArgument {
             return new StreamOffset<>(name, "$");
         }
 
+        /**
+         * Read all new arriving elements from the stream identified by {@code name} including the last element added before
+         * this call
+         *
+         * @param name must not be {@code null}.
+         * @return the {@link StreamOffset} object without a specific offset.
+         * @since 7.0
+         */
+        public static <K> StreamOffset<K> last(K name) {
+
+            LettuceAssert.notNull(name, "Stream must not be null");
+
+            return new StreamOffset<>(name, "+");
+        }
+
         /**
          * Read all new arriving elements from the stream identified by {@code name} with ids greater than the last one consumed
          * by the consumer group.


Test patch:
diff --git a/src/test/java/io/lettuce/core/commands/StreamCommandIntegrationTests.java b/src/test/java/io/lettuce/core/commands/StreamCommandIntegrationTests.java
index 2216955d4..0a4b233a9 100644
--- a/src/test/java/io/lettuce/core/commands/StreamCommandIntegrationTests.java
+++ b/src/test/java/io/lettuce/core/commands/StreamCommandIntegrationTests.java
@@ -19,25 +19,6 @@
  */
 package io.lettuce.core.commands;
 
-import static io.lettuce.core.protocol.CommandType.*;
-import static org.assertj.core.api.Assertions.*;
-
-import java.time.Duration;
-import java.time.Instant;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-
-import javax.inject.Inject;
-
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestInstance;
-import org.junit.jupiter.api.extension.ExtendWith;
-
 import io.lettuce.core.*;
 import io.lettuce.core.XReadArgs.StreamOffset;
 import io.lettuce.core.api.sync.RedisCommands;
@@ -49,6 +30,23 @@ import io.lettuce.core.output.NestedMultiOutput;
 import io.lettuce.core.protocol.CommandArgs;
 import io.lettuce.test.LettuceExtension;
 import io.lettuce.test.condition.EnabledOnCommand;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import javax.inject.Inject;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static io.lettuce.core.protocol.CommandType.XINFO;
+import static org.assertj.core.api.Assertions.assertThat;
 
 /**
  * Integration tests for {@link io.lettuce.core.api.sync.RedisStreamCommands}.
@@ -316,7 +314,7 @@ public class StreamCommandIntegrationTests extends TestSupport {
         redis.multi();
         redis.xadd("stream-1", Collections.singletonMap("key3", "value3"));
         redis.xadd("stream-2", Collections.singletonMap("key4", "value4"));
-        redis.xread(StreamOffset.from("stream-1", initial1), XReadArgs.StreamOffset.from("stream-2", initial2));
+        redis.xread(StreamOffset.from("stream-1", initial1), StreamOffset.from("stream-2", initial2));
 
         TransactionResult exec = redis.exec();
 
@@ -337,6 +335,23 @@ public class StreamCommandIntegrationTests extends TestSupport {
         assertThat(secondMessage.getBody()).containsEntry("key4", "value4");
     }
 
+    @Test
+    public void xreadLastVsLatest() {
+        redis.xadd("stream-1", Collections.singletonMap("key1", "value1"));
+        redis.xadd("stream-1", Collections.singletonMap("key2", "value2"));
+
+        List<StreamMessage<String, String>> lastMessages = redis.xread(StreamOffset.last("stream-1"));
+        List<StreamMessage<String, String>> latestMessages = redis.xread(StreamOffset.latest("stream-1"));
+
+        assertThat(lastMessages).hasSize(1);
+        StreamMessage<String, String> lastMessage = lastMessages.get(0);
+
+        assertThat(lastMessage.getStream()).isEqualTo("stream-1");
+        assertThat(lastMessage.getBody()).hasSize(1).containsEntry("key2", "value2");
+
+        assertThat(latestMessages).isEmpty();
+    }
+
     @Test
     void xinfoStream() {