Skip to content

Commit

Permalink
more flow control (#471)
Browse files Browse the repository at this point in the history
flow control example better document
  • Loading branch information
scottf authored May 17, 2021
1 parent 6ec24f9 commit 3f41101
Show file tree
Hide file tree
Showing 2 changed files with 26 additions and 5 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,23 @@ public static void main(String[] args) {
// This is configured so the subscriber ends up being considered slow
JetStreamSubscription sub = js.subscribe(subject, pso);
nc.flush(Duration.ofSeconds(5));
sub.setPendingLimits(Consumer.DEFAULT_MAX_MESSAGES, 1024);

// ------------------------------------------------------------------------------------------
// Flow Control limit is set using pending limits. You can mix and match pending messages
// with total bytes or rely on one or the other.
// ------------------------------------------------------------------------------------------
// IMPORTANT!!!! THESE VALUES ARE EXAMPLE ONLY SO THE FLOW CONTROL MESSAGES
// SHOWS UP EASILY IN THIS DEMO. IN REAL SYSTEMS YOU WILL TYPICALLY HAVE MANY MORE
// MESSAGES AND MANY MORE BYTES AS YOUR LIMITS. FOR EXAMPLE THE DEFAULT VALUES
// AS SET IN io.nats.client.Consumer ARE
// public static final long DEFAULT_MAX_MESSAGES = 64 * 1024;
// public static final long DEFAULT_MAX_BYTES = 64 * 1024 * 1024;
// ------------------------------------------------------------------------------------------
// The example first version sets pending to 500 message with 1K total bytes
// The example second version (commented out) sets pending to 1 message with 500K total bytes
// ------------------------------------------------------------------------------------------
sub.setPendingLimits(500, 1024);
// sub.setPendingLimits(1, 500000);

// publish more message data than the subscriber will handle
byte[] data = new byte[1024];
Expand Down
13 changes: 9 additions & 4 deletions src/test/java/io/nats/client/impl/JetStreamPushTests.java
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,10 @@

package io.nats.client.impl;

import io.nats.client.*;
import io.nats.client.JetStream;
import io.nats.client.JetStreamSubscription;
import io.nats.client.Message;
import io.nats.client.PushSubscribeOptions;
import io.nats.client.api.ConsumerConfiguration;
import io.nats.client.api.ConsumerInfo;
import org.junit.jupiter.api.Test;
Expand Down Expand Up @@ -294,8 +297,9 @@ public void testHeartbeat() throws Exception {
});
}

@Test
public void testFlowControl() throws Exception {
@ParameterizedTest
@ValueSource(strings = {"500,1024", "1,500000"})
public void testFlowControl(String pendingLimits) throws Exception {
runInJsServer(nc -> {
// Create our JetStream context to receive JetStream messages.
JetStream js = nc.jetStream();
Expand All @@ -312,7 +316,8 @@ public void testFlowControl() throws Exception {
// This is configured so the subscriber ends up being considered slow
JetStreamSubscription sub = js.subscribe(SUBJECT, pso);
nc.flush(Duration.ofSeconds(5));
sub.setPendingLimits(Consumer.DEFAULT_MAX_MESSAGES, 1024);
String[] split = pendingLimits.split(",");
sub.setPendingLimits(Integer.parseInt(split[0]), Integer.parseInt(split[1]));

// publish more message data than the subscriber will handle
byte[] data = new byte[1024];
Expand Down

0 comments on commit 3f41101

Please sign in to comment.