Merge branch 'main' into offset-tracking-example

Author: dc1992

.github/workflows/main.yml

@@ -21,12 +21,18 @@ jobs:
     services:
       rabbitmq:
         image: rabbitmq:3.13-rc-management
-        options: --hostname test-node
+        options: --hostname test-node --name test-node
         env:
           RABBITMQ_DEFAULT_USER: "test-user"
           RABBITMQ_DEFAULT_PASS: "test-password"
+        volumes:
+          # these directories will be empty until checkout, but they will be
+          # populated by the time we restart the service
+          - ${{ github.workspace }}/conf:/etc/rabbitmq
+          - ${{ github.workspace }}/certs:/certs
         ports:
           - 5552:5552
+          - 5551:5551
           - 5672:5672
           - 15672:15672
           - 1883:1883
@@ -41,17 +47,32 @@ jobs:
         with:
           node-version: ${{ matrix.node-version }}
           cache: "npm"
-      - name: Enable RabbitMQ Plugins
-        run: docker exec $(docker ps --filter ancestor=rabbitmq:3.13-rc-management -q) rabbitmq-plugins enable rabbitmq_stream rabbitmq_stream_management
+      - name: Generate certificates
+        env:
+          CN: test-node
+        run: |
+          git clone https://github.com/rabbitmq/tls-gen tls-gen
+          cd tls-gen/basic
+          make
+          cd ../..
+          cp -a tls-gen/basic/result certs/
+          sudo chown -R 999:999 certs
+          sudo mv certs/server_test-node_certificate.pem certs/server_rabbitmq_certificate.pem
+          sudo mv certs/server_test-node_key.pem certs/server_rabbitmq_key.pem
       - name: Restart RabbitMQ
-        run: docker restart $(docker ps --filter ancestor=rabbitmq:3.13-rc-management -q)
-      - name: Wait for rabbit instance restart
-        run: sleep 10
+        run: |
+          docker restart test-node
+          sleep 2
+          docker exec test-node rabbitmqctl await_startup
       - name: Create SuperStream
-        run: docker exec $(docker ps --filter ancestor=rabbitmq:3.13-rc-management -q) rabbitmq-streams add_super_stream super-stream-test --partitions 2
+        run: docker exec test-node rabbitmq-streams add_super_stream super-stream-test --partitions 2
       - run: npm ci
       - run: npm run check
       - run: npm run build --if-present
+      - run: |
+          docker exec test-node rabbitmqctl add_user 'O=client,CN=test-node' ''
+          docker exec test-node rabbitmqctl clear_password 'O=client,CN=test-node'
+          docker exec test-node rabbitmqctl set_permissions 'O=client,CN=test-node' '.*' '.*' '.*'
       - run: npm test
         env:
           RABBITMQ_USER: "test-user"

.gitignore

@@ -2,3 +2,4 @@ dist/
 node_modules/
 performance_test/node_modules
 .envrc
+tls-gen/

Makefile

@@ -4,6 +4,18 @@ rabbitmq-cluster:
 	mv cluster/tls-gen/basic/result/server_*_certificate.pem cluster/tls-gen/basic/result/server_certificate.pem
 	mv cluster/tls-gen/basic/result/server_*key.pem cluster/tls-gen/basic/result/server_key.pem
 	cd cluster; docker build -t haproxy-rabbitmq-cluster  .
-	cd cluster; chmod 755 -R tls-gen 
+	cd cluster; chmod -R 755 tls-gen 
 	cd cluster; docker compose down
-	cd cluster; docker compose up -d
+	cd cluster; docker compose up -d
+
+rabbitmq-test:
+	rm -rf tls-gen;
+	git clone https://github.com/rabbitmq/tls-gen tls-gen; cd tls-gen/basic; CN=rabbitmq make
+	chmod -R 755 tls-gen
+	docker compose down
+	docker compose up -d
+	sleep 5
+	docker exec rabbitmq-stream rabbitmqctl await_startup
+	docker exec rabbitmq-stream rabbitmqctl add_user 'O=client,CN=rabbitmq' ''
+	docker exec rabbitmq-stream rabbitmqctl clear_password 'O=client,CN=rabbitmq'
+	docker exec rabbitmq-stream rabbitmqctl set_permissions 'O=client,CN=rabbitmq' '.*' '.*' '.*'

README.md

@@ -374,7 +374,7 @@ npm i
 run the docker-compose to launch a rabbit instance already stream enabled
 
 ```shell
-docker-compose up -d
+make rabbitmq-test
 ```
 
 add this line to your host file (on linux `/etc/hosts`) to correctly resolve rabbitmq
@@ -400,8 +400,8 @@ npm run build
 Test:
 
 ```shell
-docker-compose up -d
-npm run test
+make rabbitmq-test
+RABBIT_MQ_TEST_NODES=rabbitmq:5552 npm run test
 ```
 
 Check everything:

conf/enabled_plugins

@@ -1 +1 @@
-[rabbitmq_management,rabbitmq_prometheus,rabbitmq_stream_management].
+[rabbitmq_management,rabbitmq_prometheus,rabbitmq_stream_management,rabbitmq_auth_mechanism_ssl].

conf/rabbitmq.conf

@@ -0,0 +1,15 @@
+loopback_users.guest = false
+
+ssl_options.cacertfile = /certs/ca_certificate.pem
+ssl_options.certfile = /certs/server_rabbitmq_certificate.pem
+ssl_options.keyfile = /certs/server_rabbitmq_key.pem
+listeners.ssl.default = 5671
+listeners.tcp.default = 5672
+stream.listeners.tcp.default = 5552
+stream.listeners.ssl.default = 5551
+auth_mechanisms.1 = PLAIN
+auth_mechanisms.2 = EXTERNAL
+ssl_options.verify               = verify_peer
+ssl_options.fail_if_no_peer_cert = false
+log.file.level = debug
+log.console = true

docker-compose.yaml

@@ -1,5 +1,3 @@
-version: "2"
-
 services:
   rabbitmq-stream:
     image: rabbitmq:3.13-rc-management
@@ -8,10 +6,13 @@ services:
     hostname: "rabbitmq"
     ports:
       - "15672:15672"
+      - "5671:5671"
       - "5672:5672"
+      - "5551:5551"
       - "5552:5552"
     environment:
       RABBITMQ_DEFAULT_USER: "rabbit"
       RABBITMQ_DEFAULT_PASS: "rabbit"
     volumes:
-      - ./conf/enabled_plugins:/etc/rabbitmq/enabled_plugins
+     - ./conf/:/etc/rabbitmq/
+     - "./tls-gen/basic/result/:/certs"

example/package-lock.json

@@ -0,0 +1,59 @@
+const rabbit = require("rabbitmq-stream-js-client")
+
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
+
+async function main() {
+  console.log("Connecting...")
+  const client = await rabbit.connect({
+    hostname: "localhost",
+    port: 5552,
+    username: "rabbit",
+    password: "rabbit",
+    vhost: "/",
+  })
+
+  console.log("Making sure the stream exists...")
+  const streamName = "stream-offset-tracking-javascript"
+  await client.createStream({ stream: streamName, arguments: {} })
+
+  const consumerRef = "offset-tracking-tutorial"
+  let firstOffset = undefined
+  let offsetSpecification = rabbit.Offset.first()
+  try {
+    const offset = await client.queryOffset({ reference: consumerRef, stream: streamName })
+    offsetSpecification = rabbit.Offset.offset(offset + 1n)
+  } catch (e) {}
+
+  let lastOffset = offsetSpecification.value
+  let messageCount = 0
+  const consumer = await client.declareConsumer(
+    { stream: streamName, offset: offsetSpecification, consumerRef },
+    async (message) => {
+      messageCount++
+      if (!firstOffset && messageCount === 1) {
+        firstOffset = message.offset
+        console.log("First message received")
+      }
+      if (messageCount % 10 === 0) {
+        await consumer.storeOffset(message.offset)
+      }
+      if (message.content.toString() === "marker") {
+        console.log("Marker found")
+        lastOffset = message.offset
+        await consumer.storeOffset(message.offset)
+        await consumer.close()
+      }
+    }
+  )
+
+  console.log(`Start consuming...`)
+  await sleep(2000)
+  console.log(`Done consuming, first offset was ${firstOffset}, last offset was ${lastOffset}`)
+}
+
+main()
+  .then(() => process.exit(0))
+  .catch((res) => {
+    console.log("Error while receiving message!", res)
+    process.exit(-1)
+  })

example/src/offset_tracking_receive.js

@@ -0,0 +1,36 @@
+const rabbit = require("rabbitmq-stream-js-client")
+
+async function main() {
+  console.log("Connecting...")
+  const client = await rabbit.connect({
+    vhost: "/",
+    port: 5552,
+    hostname: "localhost",
+    username: "rabbit",
+    password: "rabbit",
+  })
+
+  console.log("Making sure the stream exists...")
+  const streamName = "stream-offset-tracking-javascript"
+  await client.createStream({ stream: streamName, arguments: {} })
+
+  console.log("Creating the publisher...")
+  const publisher = await client.declarePublisher({ stream: streamName })
+
+  const messageCount = 100
+  console.log(`Publishing ${messageCount} messages`)
+  for (let i = 0; i < messageCount; i++) {
+    const body = i === messageCount - 1 ? "marker" : `hello ${i}`
+    await publisher.send(Buffer.from(body))
+  }
+
+  console.log("Closing the connection...")
+  await client.close()
+}
+
+main()
+  .then(() => console.log("done!"))
+  .catch((res) => {
+    console.log("Error in publishing message!", res)
+    process.exit(-1)
+  })