Java: Pub & Sub = <3 #1662

Yury-Fridlyand · 2024-06-26T06:47:47Z

Issue #, if available:
PubSub support for Java client
Ref: #1602, #1616, #1650, #1643 and #1741

Description of changes:

Configuration with callback:

    @Test
    @SneakyThrows
    public void basic_client() {
        MessageCallback callback =
            (msg, context) -> System.out.printf("Received %s, context %s\n", msg, context);

        var regularClient =
                RedisClient.CreateClient(
                                RedisClientConfiguration.builder()
                                        .address(NodeAddress.builder().port(6379).build())
                                        .requestTimeout(3000)
                                        .subscriptionConfiguration( // subscribe happens here
                                                StandaloneSubscriptionConfiguration.builder()
                                                        .subscription(EXACT, "ch1")
                                                        .subscription(PATTERN, "chat*")
                                                        .callback(callback)
                                                        .build())
                                        .build())
                        .get();

        Thread.sleep(100500);
        regularClient.close(); // unsubscribe happens here
    }

To send messages use

client.publish(channel, message).get();
clusterClient.spublish(channel, message).get();

Configuration without callback:

    @Test
    @SneakyThrows
    public void basic_client() {
        var regularClient =
                RedisClient.CreateClient(
                                RedisClientConfiguration.builder()
                                        .address(NodeAddress.builder().port(6379).build())
                                        .requestTimeout(3000)
                                        .subscriptionConfiguration(
                                                StandaloneSubscriptionConfiguration.builder()
                                                        .subscription(EXACT, "ch1")
                                                        .subscription(PATTERN, "chat*")
                                                        .build())
                                        .build())
                        .get();

        Thread.sleep(100500);

        Message msg = null;
        while ((msg = regularClient.tryGetPubSubMessage()) != null) {
            System.out.printf("Received from queue %s\n", msg);
        }
    }

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

* Add client configuartion for subscribing to channels. Signed-off-by: Yury-Fridlyand <[email protected]> * CLIPPY I HATE YOU Signed-off-by: Yury-Fridlyand <[email protected]> * Get and store callback. Signed-off-by: Yury-Fridlyand <[email protected]> * Fix tests. Signed-off-by: Yury-Fridlyand <[email protected]> * Rework configuration and add docs. Signed-off-by: Yury-Fridlyand <[email protected]> * Config rework. Signed-off-by: Yury-Fridlyand <[email protected]> * docs Signed-off-by: Yury-Fridlyand <[email protected]> * More TODOs for the god of TODOs. Signed-off-by: Yury-Fridlyand <[email protected]> --------- Signed-off-by: Yury-Fridlyand <[email protected]>

* Add `PUBLISH` and `SPUBLISH` commands. Signed-off-by: Yury-Fridlyand <[email protected]> * Fix the test. Signed-off-by: Yury-Fridlyand <[email protected]> --------- Signed-off-by: Yury-Fridlyand <[email protected]>

* Add client configuartion for subscribing to channels. Signed-off-by: Yury-Fridlyand <[email protected]> * CLIPPY I HATE YOU Signed-off-by: Yury-Fridlyand <[email protected]> * Get and store callback. Signed-off-by: Yury-Fridlyand <[email protected]> * Fix tests. Signed-off-by: Yury-Fridlyand <[email protected]> * Rework configuration and add docs. Signed-off-by: Yury-Fridlyand <[email protected]> * Config rework. Signed-off-by: Yury-Fridlyand <[email protected]> * docs Signed-off-by: Yury-Fridlyand <[email protected]> * Receive pushes (subscibed messages). Signed-off-by: Yury-Fridlyand <[email protected]> * Address PR comments. Signed-off-by: Yury-Fridlyand <[email protected]> * I HATE YOU SPOTLESS Signed-off-by: Yury-Fridlyand <[email protected]> * Rename a class. Signed-off-by: Yury-Fridlyand <[email protected]> --------- Signed-off-by: Yury-Fridlyand <[email protected]>

…bsub Signed-off-by: Yury-Fridlyand <[email protected]>

ikolomi

reviewed and posted comments from the beginning to java/client/src/main/java/glide/api/models/configuration/BaseClientConfiguration.java

java/client/src/main/java/glide/api/BaseClient.java

ikolomi · 2024-06-27T10:34:40Z

java/client/src/main/java/glide/api/BaseClient.java

+                            + " callback.");
+        }
+        // TODO
+        throw new NotImplementedException("oh no");


what is this?

Not implemented and won't be in the foreseeable future.

Can we have a more meaningful error message here, along with the TODO link to documentation (perhaps explaining the limitation?)

Updated in 5a11ff4

ikolomi · 2024-06-27T10:40:21Z

java/client/src/main/java/glide/api/BaseClient.java

-                        ? new BaseCommandResponseResolver(RedisValueResolver::valueFromPointer).apply(response)
-                        : new BaseCommandResponseResolver(RedisValueResolver::valueFromPointerBinary)
-                                .apply(response);
+                encodingUtf8 ? responseResolver.apply(response) : binaryResponseResolver.apply(response);


how its applies to pubsub?

MessageHandler uses responseResolver to get the message from received push.

Does this answer your question @ikolomi ?

ikolomi · 2024-06-27T10:41:59Z

java/client/src/main/java/glide/api/BaseClient.java

-            BaseClientConfiguration config,
-            BiFunction<ConnectionManager, CommandManager, T> constructor) {
+    protected static <T extends BaseClient> CompletableFuture<T> CreateClient(
+            @NonNull BaseClientConfiguration config, Supplier<T> constructor) {


how it is related to the pubsub

I had to rework it since BaseClient got more private fields to store.

Does this answer your question @ikolomi ?

ikolomi · 2024-06-27T11:09:26Z

java/client/src/main/java/glide/api/BaseClient.java

+        }
+        return new MessageHandler(
+                config.getSubscriptionConfiguration().getCallback(),
+                config.getSubscriptionConfiguration().getContext(),


we have a validation that context cannot be set w/o the callback and the subscriptions cannot be used with resp2, do we have it here?
https://github.com/aws/glide-for-redis/blob/e2a804c2f1d6355a797a60f28e95c2d16826a856/python/python/glide/config.py#L451

java client does not support RESP2 yet. I put a todo.

It is prohibited https://github.com/aws/glide-for-redis/blob/905a17b4fa9dec0ed3f9ad08b26437b97ba39323/java/client/src/main/java/glide/api/models/configuration/BaseSubscriptionConfiguration.java#L89-L104

@ikolomi , does this cover your concern?

Even though this takes in both callback and context in one method, the caller could supply null to either of these. I've added an explict check for this in #1773.

ikolomi · 2024-06-27T11:58:05Z

java/client/src/main/java/glide/api/models/ClusterTransaction.java

+     * @param message The message to publish.
+     * @return Command response - The number of clients that received the message.
+     */
+    public ClusterTransaction spublish(@NonNull String channel, @NonNull String message) {


very sharded

align with the reference https://github.com/aws/glide-for-redis/blob/e2a804c2f1d6355a797a60f28e95c2d16826a856/python/python/glide/async_commands/cluster_commands.py#L461

This is done in #1773

java/client/src/main/java/glide/api/models/Message.java

java/client/src/main/java/glide/api/models/configuration/BackoffStrategy.java

Signed-off-by: Yury-Fridlyand <[email protected]>

…bsub Signed-off-by: Yury-Fridlyand <[email protected]>

* Add some tests. Signed-off-by: Yury-Fridlyand <[email protected]> * Test fixes. Signed-off-by: Yury-Fridlyand <[email protected]> * Add more tests. Signed-off-by: Yury-Fridlyand <[email protected]> * Address PR comments. Signed-off-by: Yury-Fridlyand <[email protected]> * Experiment Signed-off-by: Yury-Fridlyand <[email protected]> * Add more tests. Signed-off-by: Yury-Fridlyand <[email protected]> * Typo fix. Signed-off-by: Yury-Fridlyand <[email protected]> * Typo fix. Signed-off-by: Yury-Fridlyand <[email protected]> * I HATE YOU SPOTLESS Signed-off-by: Yury-Fridlyand <[email protected]> * Typo fix. Signed-off-by: Yury-Fridlyand <[email protected]> * Uncomment test timeout. Signed-off-by: Yury-Fridlyand <[email protected]> * Address PR comments. Signed-off-by: Yury-Fridlyand <[email protected]> * Typo fix. Signed-off-by: Yury-Fridlyand <[email protected]> * Typo fix. Signed-off-by: Yury-Fridlyand <[email protected]> * Typo fix. Signed-off-by: Yury-Fridlyand <[email protected]> * Typo fix. Signed-off-by: Yury-Fridlyand <[email protected]> --------- Signed-off-by: Yury-Fridlyand <[email protected]>

…bsub Signed-off-by: Yury-Fridlyand <[email protected]>

Signed-off-by: Yury-Fridlyand <[email protected]>

java/client/src/main/java/glide/api/BaseClient.java

java/client/src/main/java/glide/api/models/BaseTransaction.java

java/client/src/main/java/glide/api/models/configuration/BaseSubscriptionConfiguration.java

java/client/src/main/java/glide/connectors/handlers/MessageHandler.java

java/client/src/main/java/glide/connectors/handlers/ReadHandler.java

jduo · 2024-06-29T22:00:40Z

java/client/src/test/java/glide/api/RedisClientCreateTest.java

@@ -45,7 +46,12 @@ public void init() {
        connectionManager = mock(ConnectionManager.class);
        threadPoolResource = mock(ThreadPoolResource.class);

-        mockedClient.when(() -> buildChannelHandler(any())).thenReturn(channelHandler);
+        RedisClient client = new RedisClient();


This client needs to close() right? Need a try-with-resources?

all instances are mocks here, so I think we can omit that.

RedisClient itself isn't a mock. Adding try-with-resources.

Nevermind, the client is now mocked.

java/client/src/test/java/glide/managers/CommandManagerTest.java

Signed-off-by: Yury-Fridlyand <[email protected]>

…bsub Signed-off-by: Yury-Fridlyand <[email protected]>

jonathanl-bq

Signing off on the Rust changes with some minor comments.

java/src/lib.rs

jonathanl-bq · 2024-06-30T16:50:47Z

java/src/lib.rs

+        // Create a java `Map<String, Object>` with two keys:
+        //   - "kind" which corresponds to the push type, stored as a `String`
+        //   - "values" which corresponds to the array of values received, stored as `Object[]`


If we only have 2 keys, why not just return an array of 2 elements? Not a big deal, but I'm curious why we went with a Map.

Good idea, but I'm following python client.

Signed-off-by: Yury-Fridlyand <[email protected]>

ikolomi

The feature is not done until all the comments are resolved and explicitly approved by myself.
Do reach out actively to me to in case of any questions arise.

java/client/src/main/java/glide/api/BaseClient.java

+        }
+        return new MessageHandler(
+                config.getSubscriptionConfiguration().getCallback(),
+                config.getSubscriptionConfiguration().getContext(),


ikolomi · 2024-07-02T15:31:00Z

java/client/src/main/java/glide/api/BaseClient.java

+                    "The operation will never complete since messages will be passed to the configured"
+                            + " callback.");
+        }
+        throw new NotImplementedException(


This has to be implemented to be on a par with python implementation

Please see #1770

Does this cover the ask @ikolomi ?

java/client/src/main/java/glide/api/BaseClient.java

java/client/src/main/java/glide/api/RedisClusterClient.java

java/client/src/main/java/glide/api/commands/PubSubClusterCommands.java

ikolomi · 2024-07-02T15:48:30Z

java/client/src/main/java/glide/api/models/ClusterTransaction.java

+     * @param message The message to publish.
+     * @return Command response - The number of clients that received the message.
+     */
+    public ClusterTransaction spublish(@NonNull String channel, @NonNull String message) {


align with the reference https://github.com/aws/glide-for-redis/blob/e2a804c2f1d6355a797a60f28e95c2d16826a856/python/python/glide/async_commands/cluster_commands.py#L461

java/client/src/main/java/glide/api/models/PubSubMessage.java

java/client/src/test/java/glide/api/models/TransactionTests.java

java/integTest/src/test/java/glide/PubSubTests.java

ikolomi · 2024-07-02T17:36:34Z

java/integTest/src/test/java/glide/PubSubTests.java

implement getPubSubMessage() flavors

implement the missing tests per reference ( https://github.com/aws/glide-for-redis/blob/main/python/python/tests/test_pubsub.py) :

test_pubsub_exact_happy_path_coexistence
test_pubsub_exact_happy_path_many_channels_co_existence
test_sharded_pubsub_co_existence
test_pubsub_pattern_co_existence
test_pubsub_pattern_many_channels
test_pubsub_combined_exact_and_pattern_one_client
test_pubsub_combined_exact_and_pattern_multiple_clients
test_pubsub_combined_exact_pattern_and_sharded_multi_client
test_pubsub_combined_different_channels_with_same_name
test_pubsub_two_publishing_clients_same_name
test_pubsub_exact_max_size_message
test_pubsub_sharded_max_size_message
test_pubsub_exact_max_size_message_callback
test_pubsub_sharded_max_size_message_callback
test_pubsub_resp2_raise_an_error
test_pubsub_context_with_no_callback_raise_error

All "coexistence" tests depend on getPubSubMessage and wait for it.

All "max_size_message" tests disabled on python side due to a bug and there is no reason to implement them now until bug fixed.

RESP2 is not supported by java client - no test there.

All rest are implemented.

eifrah-aws · 2024-07-03T13:14:33Z

java/client/src/main/java/glide/api/BaseClient.java

+    public CompletableFuture<String> publish(@NonNull String channel, @NonNull String message) {
+        return commandManager.submitNewCommand(
+                Publish,
+                new String[] {channel, message},


why are we still using Strings?

eifrah-aws · 2024-07-03T13:15:25Z

java/client/src/main/java/glide/api/RedisClusterClient.java

@@ -780,6 +782,18 @@ public CompletableFuture<String> randomKey() {
                RandomKey, new String[0], this::handleStringOrNullResponse);
    }

+    @Override
+    public CompletableFuture<String> spublish(@NonNull String channel, @NonNull String message) {


Please add a GlideString variant for this function

@eifrah-aws , I've removed spublish and made two overloads of publish(). They both use ArgType now. Please see: #1773.

eifrah-aws · 2024-07-03T13:15:42Z

java/client/src/main/java/glide/api/commands/PubSubClusterCommands.java

+     * assert response.equals("OK");
+     * }</pre>
+     */
+    CompletableFuture<String> spublish(String channel, String message);


GlideString variant

eifrah-aws · 2024-07-03T13:15:56Z

java/client/src/main/java/glide/api/commands/PubSubBaseCommands.java

+     * assert response.equals("OK");
+     * }</pre>
+     */
+    CompletableFuture<String> publish(String channel, String message);


GlideString variant

* Java: Add client configuration for subscribing to channels. (#381) * Add client configuartion for subscribing to channels. Signed-off-by: Yury-Fridlyand <[email protected]> * CLIPPY I HATE YOU Signed-off-by: Yury-Fridlyand <[email protected]> * Get and store callback. Signed-off-by: Yury-Fridlyand <[email protected]> * Fix tests. Signed-off-by: Yury-Fridlyand <[email protected]> * Rework configuration and add docs. Signed-off-by: Yury-Fridlyand <[email protected]> * Config rework. Signed-off-by: Yury-Fridlyand <[email protected]> * docs Signed-off-by: Yury-Fridlyand <[email protected]> * More TODOs for the god of TODOs. Signed-off-by: Yury-Fridlyand <[email protected]> --------- Signed-off-by: Yury-Fridlyand <[email protected]> * Add `PUBLISH` and `SPUBLISH` commands. (#391) * Add `PUBLISH` and `SPUBLISH` commands. Signed-off-by: Yury-Fridlyand <[email protected]> * Fix the test. Signed-off-by: Yury-Fridlyand <[email protected]> --------- Signed-off-by: Yury-Fridlyand <[email protected]> * Java client: receive pubsub messages (#385) * Add client configuartion for subscribing to channels. Signed-off-by: Yury-Fridlyand <[email protected]> * CLIPPY I HATE YOU Signed-off-by: Yury-Fridlyand <[email protected]> * Get and store callback. Signed-off-by: Yury-Fridlyand <[email protected]> * Fix tests. Signed-off-by: Yury-Fridlyand <[email protected]> * Rework configuration and add docs. Signed-off-by: Yury-Fridlyand <[email protected]> * Config rework. Signed-off-by: Yury-Fridlyand <[email protected]> * docs Signed-off-by: Yury-Fridlyand <[email protected]> * Receive pushes (subscibed messages). Signed-off-by: Yury-Fridlyand <[email protected]> * Address PR comments. Signed-off-by: Yury-Fridlyand <[email protected]> * I HATE YOU SPOTLESS Signed-off-by: Yury-Fridlyand <[email protected]> * Rename a class. Signed-off-by: Yury-Fridlyand <[email protected]> --------- Signed-off-by: Yury-Fridlyand <[email protected]> * Address PR comments. Signed-off-by: Yury-Fridlyand <[email protected]> * Address PR comments. Signed-off-by: Yury-Fridlyand <[email protected]> * Java: add IT for pubsub (#400) * Add some tests. Signed-off-by: Yury-Fridlyand <[email protected]> * Test fixes. Signed-off-by: Yury-Fridlyand <[email protected]> * Add more tests. Signed-off-by: Yury-Fridlyand <[email protected]> * Address PR comments. Signed-off-by: Yury-Fridlyand <[email protected]> * Experiment Signed-off-by: Yury-Fridlyand <[email protected]> * Add more tests. Signed-off-by: Yury-Fridlyand <[email protected]> * Typo fix. Signed-off-by: Yury-Fridlyand <[email protected]> * Typo fix. Signed-off-by: Yury-Fridlyand <[email protected]> * I HATE YOU SPOTLESS Signed-off-by: Yury-Fridlyand <[email protected]> * Typo fix. Signed-off-by: Yury-Fridlyand <[email protected]> * Uncomment test timeout. Signed-off-by: Yury-Fridlyand <[email protected]> * Address PR comments. Signed-off-by: Yury-Fridlyand <[email protected]> * Typo fix. Signed-off-by: Yury-Fridlyand <[email protected]> * Typo fix. Signed-off-by: Yury-Fridlyand <[email protected]> * Typo fix. Signed-off-by: Yury-Fridlyand <[email protected]> * Typo fix. Signed-off-by: Yury-Fridlyand <[email protected]> --------- Signed-off-by: Yury-Fridlyand <[email protected]> * Update function signature. Signed-off-by: Yury-Fridlyand <[email protected]> * Address PR comments. Signed-off-by: Yury-Fridlyand <[email protected]> * Address PR comments. Signed-off-by: Yury-Fridlyand <[email protected]> --------- Signed-off-by: Yury-Fridlyand <[email protected]>

Yury-Fridlyand added 4 commits June 25, 2024 23:09

Add PUBLISH and SPUBLISH commands. (#391)

892bf2b

* Add `PUBLISH` and `SPUBLISH` commands. Signed-off-by: Yury-Fridlyand <[email protected]> * Fix the test. Signed-off-by: Yury-Fridlyand <[email protected]> --------- Signed-off-by: Yury-Fridlyand <[email protected]>

Merge remote-tracking branch 'upstream/main' into java/integ_yuryf_pu…

208c943

…bsub Signed-off-by: Yury-Fridlyand <[email protected]>

Yury-Fridlyand added the java issues and fixes related to the java client label Jun 26, 2024

Yury-Fridlyand requested review from aaron-congo, jonathanl-bq, acarbonetto and ikolomi June 26, 2024 06:47

Yury-Fridlyand requested a review from a team as a code owner June 26, 2024 06:47

Yury-Fridlyand requested review from GumpacG, barshaul and shohamazon June 26, 2024 06:48

Yury-Fridlyand marked this pull request as draft June 26, 2024 06:48

ikolomi reviewed Jun 27, 2024

View reviewed changes

Yury-Fridlyand added 6 commits June 28, 2024 21:32

Address PR comments.

27d789c

Signed-off-by: Yury-Fridlyand <[email protected]>

Address PR comments.

5a11ff4

Signed-off-by: Yury-Fridlyand <[email protected]>

Merge remote-tracking branch 'upstream/main' into java/integ_yuryf_pu…

b43eee1

…bsub Signed-off-by: Yury-Fridlyand <[email protected]>

Merge remote-tracking branch 'upstream/main' into java/integ_yuryf_pu…

bd3468b

…bsub Signed-off-by: Yury-Fridlyand <[email protected]>

Update function signature.

bf97591

Signed-off-by: Yury-Fridlyand <[email protected]>

Yury-Fridlyand changed the title ~~[WIP] Java: Pub & Sub = <3~~ Java: Pub & Sub = <3 Jun 29, 2024

Yury-Fridlyand marked this pull request as ready for review June 29, 2024 20:21

jduo suggested changes Jun 29, 2024

View reviewed changes

Yury-Fridlyand added 2 commits June 29, 2024 16:47

Address PR comments.

098cc43

Signed-off-by: Yury-Fridlyand <[email protected]>

Merge remote-tracking branch 'upstream/main' into java/integ_yuryf_pu…

51c0111

…bsub Signed-off-by: Yury-Fridlyand <[email protected]>

jonathanl-bq approved these changes Jun 30, 2024

View reviewed changes

jduo approved these changes Jun 30, 2024

View reviewed changes

Address PR comments.

6e1ca63

Signed-off-by: Yury-Fridlyand <[email protected]>

acarbonetto approved these changes Jul 1, 2024

View reviewed changes

acarbonetto merged commit eb2201c into valkey-io:main Jul 1, 2024
17 checks passed

acarbonetto deleted the java/integ_yuryf_pubsub branch July 1, 2024 03:29

Yury-Fridlyand mentioned this pull request Jul 1, 2024

Java: Add Logger #1422

Merged

ikolomi reviewed Jul 2, 2024

View reviewed changes

Yury-Fridlyand mentioned this pull request Jul 2, 2024

Java: async getPubSubMessage. #1770

Closed

3 tasks

jduo mentioned this pull request Jul 3, 2024

Java: Address pubsub PR comments #1773

Merged

eifrah-aws reviewed Jul 3, 2024

View reviewed changes

Java: Pub & Sub = <3 #1662

Java: Pub & Sub = <3 #1662

Conversation

Yury-Fridlyand commented Jun 26, 2024 • edited Loading

Configuration with callback:

Configuration without callback:

ikolomi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Yury-Fridlyand Jun 29, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

This comment was marked as resolved.

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jonathanl-bq left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ikolomi left a comment

Choose a reason for hiding this comment

This comment was marked as resolved.

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Yury-Fridlyand commented Jun 26, 2024 •

edited

Loading

Yury-Fridlyand Jun 29, 2024 •

edited

Loading