Skip to content

Commit

Permalink
updated readme gradle and maven for snapshots (#539)
Browse files Browse the repository at this point in the history
  • Loading branch information
@scottf
scottf authored Sep 8, 2021
1 parent a8a17c9 commit fd54e70
Show file tree
Hide file tree
Showing 3 changed files with 133 additions and 42 deletions.
40 changes: 38 additions & 2 deletions README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -101,10 +101,22 @@ repositories {
maven {
url "https://oss.sonatype.org/content/repositories/releases"
}
}
```

If you need a snapshot version, you must add the url for the snapshots and change your dependency.

```groovy
repositories {
...
maven {
url "https://oss.sonatype.org/content/repositories/snapshots"
}
}
dependencies {
implementation 'io.nats:jnats:2.12.0-SNAPSHOT'
}
```

### Using Maven
Expand All @@ -126,10 +138,34 @@ If you need the absolute latest, before it propagates to maven central, you can
<repository>
<id>latest-repo</id>
<url>https://oss.sonatype.org/content/repositories/releases</url>
<releases><enabled>true</enabled></releases>
<snapshots><enabled>false</enabled></snapshots>
<releases>
<enabled>true</enabled>
</releases>
</repository>
</repositories>
```

If you need a snapshot version, you must enable snapshots and change your dependency.

```xml
<repositories>
<repository>
<id>latest-repo</id>
<url>https://oss.sonatype.org/content/repositories/releases</url>
<releases>
<enabled>true</enabled>
</releases>
<snapshots>
<enabled>true</enabled>
</snapshots>
</repository>
</repositories>

<dependency>
<groupId>io.nats</groupId>
<artifactId>jnats</artifactId>
<version>2.12.0-SNAPSHOT</version>
</dependency>
```

If you are using the 1.x version of java-nats and don't want to upgrade to 2.0.0 please use ranges in your POM file, java-nats-streaming 1.x is using [1.1, 1.9.9) for this.
Expand Down
107 changes: 73 additions & 34 deletions src/examples/java/io/nats/examples/jetstream/NatsJsPullSubExpireUseCases.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,10 @@ public static void main(String[] args) {
try (Connection nc = Nats.connect(ExampleUtils.createExampleOptions(exArgs.server))) {
NatsJsUtils.createOrUpdateStream(nc, exArgs.stream, exArgs.subject);

// after version 2.4.0 the server behavior changed regarding returning 408's
// and that it no longer returns any 408s
boolean no408sAreComing = nc.getServerInfo().isNewerVersionThan("2.4.0");

// Create our JetStream context to receive JetStream messages.
JetStream js = nc.jetStream();

Expand All @@ -66,71 +70,106 @@ public static void main(String[] args) {
JetStreamSubscription sub = js.subscribe(exArgs.subject, pullOptions);
nc.flush(Duration.ofSeconds(1));

sub.pullExpiresIn(10, Duration.ofSeconds(2));

// 1. Publish some that is less than the batch size.
System.out.println("\n----------\n1. Publish some amount of messages, but not entire batch size.");
publish(js, exArgs.subject, "A", 6);
sub.pullExpiresIn(10, Duration.ofMillis(1200));
List<Message> messages = readMessagesAck(sub, Duration.ofSeconds(2));
System.out.println("We should have received 6 total messages, we received: " + messages.size());
System.out.println("We should have received 6 regular messages, we received: " + countJs(messages));
System.out.println("We should have received 0 408 markers of last batch, we received: " + count408s(messages));
if (no408sAreComing) {
System.out.println("We should have received 6 total messages, we received: " + messages.size());
}
else {
System.out.println("We should have received 6 total messages, we received: " + messages.size());
System.out.println("We should have received 6 regular messages, we received: " + countJs(messages));
System.out.println("We should have received 0 408 markers of last batch, we received: " + count408s(messages));
}

// 2. Publish some more covering our pull size...
System.out.println("----------\n2. Publish more than the batch size.");
sub.pullExpiresIn(10, Duration.ofSeconds(2));
sub.pullExpiresIn(10, Duration.ofMillis(1200));
publish(js, exArgs.subject, "B", 14);
messages = readMessagesAck(sub, Duration.ofSeconds(2));
System.out.println("We should have received 14 total messages, we received: " + messages.size());
System.out.println("We should have received 10 regular messages, we received: " + countJs(messages));
System.out.println("We should have received 4 408 markers of last batch, we received: " + count408s(messages));
if (no408sAreComing) {
System.out.println("We should have received 10 total messages, we received: " + messages.size());
}
else {
System.out.println("We should have received 14 total messages, we received: " + messages.size());
System.out.println("We should have received 10 regular messages, we received: " + countJs(messages));
System.out.println("We should have received 4 408 markers of last batch, we received: " + count408s(messages));
}

// 3. There are still 4 messages from B, but the batch was finished
// - won't get any messages until a pull is issued.
System.out.println("----------\n3. Read without re-issue.");
System.out.println("----------\n3. Read without issuing a pull.");
messages = readMessagesAck(sub, Duration.ofSeconds(2));
System.out.println("We should have received 0 total messages, we received: " + messages.size());

// 4. re-issue the pull to get the last 4
System.out.println("----------\n4. Re-issue to get the last 4.");
sub.pullExpiresIn(10, Duration.ofSeconds(2));
System.out.println("----------\n4. Issue the pull to get the last 4.");
sub.pullExpiresIn(10, Duration.ofMillis(1200));
messages = readMessagesAck(sub, Duration.ofSeconds(2));
System.out.println("We should have received 4 total messages, we received: " + messages.size());
System.out.println("We should have received 4 regular messages, we received: " + countJs(messages));
System.out.println("We should have received 0 408 markers of last batch, we received: " + count408s(messages));
if (no408sAreComing) {
System.out.println("We should have received 4 total messages, we received: " + messages.size());
}
else {
System.out.println("We should have received 4 total messages, we received: " + messages.size());
System.out.println("We should have received 4 regular messages, we received: " + countJs(messages));
System.out.println("We should have received 0 408 markers of last batch, we received: " + count408s(messages));
}

// 5. publish a lot of messages
System.out.println("----------\n5. Publish a lot of messages. The last pull was under the batch size.");
System.out.println( " Issue another pull with batch size less than number of messages.");
publish(js, exArgs.subject, "C", 25);
sub.pullExpiresIn(10, Duration.ofSeconds(2));
sub.pullExpiresIn(10, Duration.ofMillis(1200));
messages = readMessagesAck(sub, Duration.ofSeconds(2));
System.out.println("We should have received 16 total messages, we received: " + messages.size());
System.out.println("We should have received 10 regular messages, we received: " + countJs(messages));
System.out.println("We should have received 6 408 markers of last batch, we received: " + count408s(messages));
if (no408sAreComing) {
System.out.println("We should have received 10 total messages, we received: " + messages.size());
}
else {
System.out.println("We should have received 16 total messages, we received: " + messages.size());
System.out.println("We should have received 10 regular messages, we received: " + countJs(messages));
System.out.println("We should have received 6 408 markers of last batch, we received: " + count408s(messages));
}

// 6. there are still more messages
System.out.println("----------\n6. Still more messages.");
sub.pullExpiresIn(10, Duration.ofSeconds(2));
System.out.println("----------\n6. Still more messages. Issue another pull with batch size less than number of messages.");
sub.pullExpiresIn(10, Duration.ofMillis(1200));
messages = readMessagesAck(sub, Duration.ofSeconds(2));
System.out.println("We should have received 1 total messages, we received: " + messages.size());
System.out.println("We should have received 10 regular messages, we received: " + countJs(messages));
System.out.println("We should have received 0 408 markers of last batch, we received: " + count408s(messages));
if (no408sAreComing) {
System.out.println("We should have received 10 total messages, we received: " + messages.size());
}
else {
System.out.println("We should have received 10 total messages, we received: " + messages.size());
System.out.println("We should have received 10 regular messages, we received: " + countJs(messages));
System.out.println("We should have received 0 408 markers of last batch, we received: " + count408s(messages));
}

// 7. there are still more messages
System.out.println("----------\n7. Still more messages.");
sub.pullExpiresIn(10, Duration.ofSeconds(2));
System.out.println("----------\n7. Still more messages. Issue another pull with batch size more than number of messages.");
sub.pullExpiresIn(10, Duration.ofMillis(1200));
messages = readMessagesAck(sub, Duration.ofSeconds(2));
System.out.println("We should have received 5 total messages, we received: " + messages.size());
System.out.println("We should have received 5 regular messages, we received: " + countJs(messages));
System.out.println("We should have received 0 408 markers of last batch, we received: " + count408s(messages));
if (no408sAreComing) {
System.out.println("We should have received 5 total messages, we received: " + messages.size());
}
else {
System.out.println("We should have received 5 total messages, we received: " + messages.size());
System.out.println("We should have received 5 regular messages, we received: " + countJs(messages));
System.out.println("We should have received 0 408 markers of last batch, we received: " + count408s(messages));
}

// 8. we got them all
System.out.println("----------\n7. No messages left.");
sub.pullExpiresIn(10, Duration.ofSeconds(2));
System.out.println("----------\n8. No messages left.");
sub.pullExpiresIn(10, Duration.ofMillis(1200));
messages = readMessagesAck(sub, Duration.ofSeconds(2));
System.out.println("We should have received 5 total messages, we received: " + messages.size());
System.out.println("We should have received 0 regular messages, we received: " + countJs(messages));
System.out.println("We should have received 5 408 markers of last batch, we received: " + count408s(messages));
if (no408sAreComing) {
System.out.println("We should have received 0 total messages, we received: " + messages.size());
}
else {
System.out.println("We should have received 5 total messages, we received: " + messages.size());
System.out.println("We should have received 0 regular messages, we received: " + countJs(messages));
System.out.println("We should have received 5 408 markers of last batch, we received: " + count408s(messages));
}

System.out.println("----------\n");
}
Expand Down
28 changes: 22 additions & 6 deletions src/test/java/io/nats/client/impl/JetStreamPullTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -318,6 +318,7 @@ public void testAfterIncompleteExpiresPulls() throws Exception {
runInJsServer(nc -> {
// Create our JetStream context to receive JetStream messages.
JetStream js = nc.jetStream();
boolean serverHasExpireChange = nc.getServerInfo().isNewerVersionThan("2.4.0");

// create the stream.
createMemoryStream(nc, STREAM, SUBJECT);
Expand All @@ -343,8 +344,13 @@ public void testAfterIncompleteExpiresPulls() throws Exception {
jsPublish(js, SUBJECT, "B", 10);
sub.pullExpiresIn(10, Duration.ofMillis(expires)); // using Duration version here
messages = readMessagesAck(sub);
assertEquals(15, messages.size());
assertStarts408(messages, 5, 10);
if (serverHasExpireChange) {
assertEquals(10, messages.size());
}
else {
assertEquals(15, messages.size());
assertStarts408(messages, 5, 10);
}
sleep(expires); // make sure the pull actually expires

jsPublish(js, SUBJECT, "C", 5);
Expand All @@ -357,8 +363,13 @@ public void testAfterIncompleteExpiresPulls() throws Exception {
jsPublish(js, SUBJECT, "D", 10);
sub.pull(10);
messages = readMessagesAck(sub);
assertEquals(15, messages.size());
assertStarts408(messages, 5, 10);
if (serverHasExpireChange) {
assertEquals(10, messages.size());
}
else {
assertEquals(15, messages.size());
assertStarts408(messages, 5, 10);
}

jsPublish(js, SUBJECT, "E", 5);
sub.pullExpiresIn(10, expires); // using millis version here
Expand All @@ -370,8 +381,13 @@ public void testAfterIncompleteExpiresPulls() throws Exception {
jsPublish(js, SUBJECT, "F", 10);
sub.pullNoWait(10);
messages = readMessagesAck(sub);
assertEquals(15, messages.size());
assertStarts408(messages, 5, 10);
if (serverHasExpireChange) {
assertEquals(10, messages.size());
}
else {
assertEquals(15, messages.size());
assertStarts408(messages, 5, 10);
}

jsPublish(js, SUBJECT, "G", 5);
sub.pullExpiresIn(10, expires); // using millis version here
Expand Down

0 comments on commit fd54e70

Please sign in to comment.