Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

add support for JetStream #254

Merged
merged 21 commits into from
Dec 5, 2022
Merged

add support for JetStream #254

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
c9c8ba5
simplify configuration of nats
grimmjo Jul 1, 2022
c8a67bd
adds jetstream configuration and factory
grimmjo Jul 1, 2022
73bd536
adds publishing support for jetstream
grimmjo Jul 26, 2022
81ef93b
disable japiCmp for now
grimmjo Jul 26, 2022
1b88e22
adds support for push consumers
grimmjo Jul 27, 2022
0737efc
include jetstream module into nats module
grimmjo Aug 1, 2022
5ee5a1b
update jnats to 2.16.4
grimmjo Oct 6, 2022
379fe94
rework JetStreamFactory to eager initialize streams
grimmjo Oct 7, 2022
6e507ed
rework push consumers
grimmjo Oct 7, 2022
09c30f5
adds message acknowledging for push consumers
grimmjo Oct 7, 2022
b366cf4
adds push and pull consumer registry
grimmjo Oct 10, 2022
43663e1
introduce StaticConsumerState
grimmjo Oct 10, 2022
895da24
fixes sonar issues
grimmjo Oct 10, 2022
c480e10
adds default values for push consumer annotation
grimmjo Oct 10, 2022
d2fbc55
fixes docs-examples tests
grimmjo Oct 10, 2022
57cc3f9
extends docs-examples test for jetstream
grimmjo Oct 10, 2022
91373b3
fixes unit test error
grimmjo Oct 10, 2022
5065029
adds documentation for jetstream
grimmjo Oct 11, 2022
77d3aa3
adds ordered push consumer support
grimmjo Oct 12, 2022
9eb628f
fixes some small code review issues
grimmjo Nov 15, 2022
7bb1845
Fix issues after upgrading to micronaut 4.0.0-SNAPSHOT
grimmjo Nov 29, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
250 changes: 249 additions & 1 deletion config/accepted-api-changes.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,5 +48,253 @@
"type": "io.micronaut.nats.serdes.JsonNatsMessageSerDes",
"member": "Constructor io.micronaut.nats.serdes.JsonNatsMessageSerDes(com.fasterxml.jackson.databind.ObjectMapper)",
"reason": "Migration to Micronaut 4.0.0"
}
},
{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Class io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Field $ANNOTATION_METADATA",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Method io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference.getBeanDefinitionType()",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Method io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference.getBeanType()",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Method io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference.load()",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Constructor io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference()",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$Definition$Reference",
"member": "Class io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$Definition$Reference",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$Definition$Reference",
"member": "Field $ANNOTATION_METADATA",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$Definition$Reference",
"member": "Method io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$Definition$Reference.getBeanDefinitionType()",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$Definition$Reference",
"member": "Method io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$Definition$Reference.getBeanType()",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$Definition$Reference",
"member": "Method io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$Definition$Reference.load()",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$Definition$Reference",
"member": "Constructor io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$Definition$Reference()",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.SingleNatsConnectionFactoryConfig",
"member": "Class io.micronaut.nats.connect.SingleNatsConnectionFactoryConfig",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.SingleNatsConnectionFactoryConfig",
"member": "Field DEFAULT_NAME",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.SingleNatsConnectionFactoryConfig",
"member": "Method io.micronaut.nats.connect.SingleNatsConnectionFactoryConfig.setTls(io.micronaut.nats.connect.SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration)",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.SingleNatsConnectionFactoryConfig",
"member": "Constructor io.micronaut.nats.connect.SingleNatsConnectionFactoryConfig()",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration",
"member": "Class io.micronaut.nats.connect.SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration",
"member": "Constructor io.micronaut.nats.connect.SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration()",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration",
"member": "Class io.micronaut.nats.connect.ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration",
"member": "Constructor io.micronaut.nats.connect.ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration()",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.ClusterNatsConnectionFactoryConfig",
"member": "Class io.micronaut.nats.connect.ClusterNatsConnectionFactoryConfig",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.ClusterNatsConnectionFactoryConfig",
"member": "Field PREFIX",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.ClusterNatsConnectionFactoryConfig",
"member": "Method io.micronaut.nats.connect.ClusterNatsConnectionFactoryConfig.setTls(io.micronaut.nats.connect.ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration)",
"reason": "Provide nats configurations in a generic way"
},
{
"type": "io.micronaut.nats.connect.ClusterNatsConnectionFactoryConfig",
"member": "Constructor io.micronaut.nats.connect.ClusterNatsConnectionFactoryConfig(java.lang.String)",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Class io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Field $ANNOTATION_METADATA",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Method io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference.getBeanDefinitionType()",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Method io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference.getBeanType()",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Method io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference.load()",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Constructor io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference()",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$Definition$Reference",
"member": "Class io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$Definition$Reference",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$Definition$Reference",
"member": "Field $ANNOTATION_METADATA",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$Definition$Reference",
"member": "Method io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$Definition$Reference.getBeanDefinitionType()",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$Definition$Reference",
"member": "Method io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$Definition$Reference.getBeanType()",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$Definition$Reference",
"member": "Method io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$Definition$Reference.load()",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$Definition$Reference",
"member": "Constructor io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$Definition$Reference()",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Implemented interface io.micronaut.core.annotation.AnnotationSource",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Implemented interface io.micronaut.inject.BeanContextConditional",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Implemented interface io.micronaut.core.annotation.AnnotationMetadataProvider",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Implemented interface io.micronaut.inject.BeanType",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Implemented interface io.micronaut.inject.BeanDefinitionReference",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$Definition$Reference",
"member": "Implemented interface io.micronaut.core.annotation.AnnotationSource",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$Definition$Reference",
"member": "Implemented interface io.micronaut.inject.BeanContextConditional",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$Definition$Reference",
"member": "Implemented interface io.micronaut.core.annotation.AnnotationMetadataProvider",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$Definition$Reference",
"member": "Implemented interface io.micronaut.inject.BeanType",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$ClusterNatsConnectionFactoryConfig$Definition$Reference",
"member": "Implemented interface io.micronaut.inject.BeanDefinitionReference",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Implemented interface io.micronaut.core.annotation.AnnotationSource",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Implemented interface io.micronaut.inject.BeanContextConditional",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Implemented interface io.micronaut.core.annotation.AnnotationMetadataProvider",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Implemented interface io.micronaut.inject.BeanType",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$DefaultTlsConfiguration$Definition$Reference",
"member": "Implemented interface io.micronaut.inject.BeanDefinitionReference",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$Definition$Reference",
"member": "Implemented interface io.micronaut.core.annotation.AnnotationSource",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$Definition$Reference",
"member": "Implemented interface io.micronaut.inject.BeanContextConditional",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$Definition$Reference",
"member": "Implemented interface io.micronaut.core.annotation.AnnotationMetadataProvider",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$Definition$Reference",
"member": "Implemented interface io.micronaut.inject.BeanType",
"reason": "Provide nats configurations in a generic way"
},{
"type": "io.micronaut.nats.connect.$SingleNatsConnectionFactoryConfig$Definition$Reference",
"member": "Implemented interface io.micronaut.inject.BeanDefinitionReference",
"reason": "Provide nats configurations in a generic way"
}
]
8 changes: 8 additions & 0 deletions docs-examples/example-groovy/build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,15 @@ dependencies {
testImplementation mn.reactor
testImplementation mn.micronaut.test.spock
testImplementation mn.spock
testRuntimeOnly libs.junit.jupiter.engine
}

test {
useJUnitPlatform()

jvmArgs '-Duser.country=US'
jvmArgs '-Duser.language=en'
failFast = true
}

tasks.withType(GroovyCompile) {
Expand Down
7 changes: 5 additions & 2 deletions docs-examples/example-groovy/src/test/groovy/io/micronaut/nats/AbstractNatsTest.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ abstract class AbstractNatsTest extends Specification {
static GenericContainer natsContainer =
new GenericContainer("nats:latest")
.withExposedPorts(4222)
.withCommand("--js")
.waitingFor(new LogMessageWaitStrategy().withRegEx("(?s).*Server is ready.*"))

static {
Expand All @@ -25,8 +26,10 @@ abstract class AbstractNatsTest extends Specification {
}

protected Map<String, Object> getConfiguration() {
["nats.addresses": "nats://localhost:" + natsContainer.getMappedPort(4222),
"spec.name": getClass().simpleName] as Map
["nats.default.addresses": "nats://localhost:" + natsContainer.getMappedPort(4222),
"spec.name" : getClass().simpleName,
"nats.default.jetstream.streams.events.storage-type": "Memory",
"nats.default.jetstream.streams.events.subjects": ["events.>"]] as Map
}

protected void waitFor(Closure<?> conditionEvaluator) {
Expand Down
2 changes: 1 addition & 1 deletion ...e-groovy/src/test/groovy/io/micronaut/nats/docs/consumer/connection/ConnectionSpec.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,6 @@ class ConnectionSpec extends AbstractNatsTest {
}

protected Map<String, Object> getConfiguration() {
super.configuration + ["nats.servers.product-cluster.addresses": super.configuration.remove("nats.addresses")]
super.configuration + ["nats.product-cluster.addresses": super.configuration.remove("nats.default.addresses")]
}
}
78 changes: 78 additions & 0 deletions ...ples/example-groovy/src/test/groovy/io/micronaut/nats/docs/jetstream/JetstreamSpec.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,78 @@
package io.micronaut.nats.docs.jetstream

import io.micronaut.nats.AbstractNatsTest
import io.nats.client.JetStreamManagement
import io.nats.client.PublishOptions
import io.nats.client.api.PublishAck

import java.nio.charset.StandardCharsets

class JetstreamSpec extends AbstractNatsTest {

void "simple producer"() {
startContext()

when:
// tag::producer[]
def productClient = applicationContext.getBean(ProductClient)
PublishAck pa = productClient.send("events.one", "ghi".getBytes(StandardCharsets.UTF_8),
PublishOptions.builder()
.stream("events")
.messageId("id00001")
.expectedStream("events")
.build())
productClient.send("events.two", "jkl".getBytes(StandardCharsets.UTF_8),
PublishOptions.builder()
.stream("events")
.messageId("id00002")
.expectedStream("events")
.expectedLastMsgId("id00001")
.expectedLastSequence(pa.getSeqno())
.build())
// end::producer[]

ProductListener productListener = applicationContext.getBean(ProductListener)
JetStreamManagement jsm = applicationContext.getBean(JetStreamManagement)

then:
waitFor {
productListener.messageLengths.size() == 2
jsm.getStreamInfo("events").getStreamState().getMsgCount() == 2
}

cleanup:
jsm.deleteStream("events")
}

void "pull consumer"() {
startContext()

when:
def productClient = applicationContext.getBean(ProductClient)
PublishAck pa = productClient.send("events.three", "ghi".getBytes(StandardCharsets.UTF_8),
PublishOptions.builder()
.stream("events")
.messageId("id00001")
.expectedStream("events")
.build())
productClient.send("events.four", "jkl".getBytes(StandardCharsets.UTF_8),
PublishOptions.builder()
.stream("events")
.messageId("id00002")
.expectedStream("events")
.expectedLastMsgId("id00001")
.expectedLastSequence(pa.getSeqno())
.build())

PullConsumerHelper pullConsumerHelper = applicationContext.getBean(PullConsumerHelper)

then:
waitFor {
pullConsumerHelper.pullMessages().size() == 2
}

cleanup:
JetStreamManagement jsm = applicationContext.getBean(JetStreamManagement)
jsm.deleteStream("events")
}
}
Loading