NATS JetStream + Go

1.txt

$docker-compose up

[+] Running 1/0
 ⠿ Container nat-js-nat-js-1  Created                                                  0.0s
Attaching to nat-js-nat-js-1
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.859286 [INF] Starting nats-server
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.859416 [INF]   Version:  2.8.4
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.859428 [INF]   Git:      [66524ed]
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.859442 [INF]   Name:     NBYAJSZUVNQHW4VUQZC2JDJX4QPECWSQTNJSXZAYKLL3IGTOFXART7FG
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.859460 [INF]   Node:     a7Zbpbhh
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.859483 [INF]   ID:       NBYAJSZUVNQHW4VUQZC2JDJX4QPECWSQTNJSXZAYKLL3IGTOFXART7FG
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.862737 [INF] Starting JetStream
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.863645 [INF]     _ ___ _____ ___ _____ ___ ___   _   __  __
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.864178 [INF]  _ | | __|_   _/ __|_   _| _ \ __| /_\ |  \/  |
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.864308 [INF] | || | _|  | | \__ \ | | |   / _| / _ \| |\/| |
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.864381 [INF]  \__/|___| |_| |___/ |_| |_|_\___/_/ \_\_|  |_|
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.864438 [INF] 
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.864494 [INF]          https://docs.nats.io/jetstream
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.864552 [INF] 
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.864754 [INF] ---------------- JETSTREAM ----------------
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.864895 [INF]   Max Memory:      4.35 GB
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.865038 [INF]   Max Storage:     33.31 GB
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.865139 [INF]   Store Directory: "/tmp/nats/jetstream"
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.865285 [INF] -------------------------------------------
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.884144 [INF]   Restored 5 messages for stream '$G > demo'
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.888166 [INF] Listening for client connections on 0.0.0.0:4222
nat-js-nat-js-1  | [1] 2022/07/06 11:32:04.890184 [INF] Server is ready

connect.go

package demo

import (
	"github.com/nats-io/nats.go"
)

func InitialNatServer() (nats.JetStreamContext, error) {
	n, err := nats.Connect("nats://127.0.0.1:4222")
	if err != nil {
		return nil, err
	}
	context, err := n.JetStream(nats.PublishAsyncMaxPending(256))
	if err != nil {
		return nil, err
	}

	s, err := context.StreamInfo("demo")
	if s == nil {
		// Create new stream
		_, err := context.AddStream(&nats.StreamConfig{
			Name:     "demo",
			Subjects: []string{"demo.*"},
		})
		if err != nil {
			return nil, err
		}
	}

	return context, nil
}

docker-compose.yml

version: "3"

services:
  nat-js:
    image: nats:2.8.4-alpine3.15
    command: "-js"
    ports:
      - 4222:4222

pub.go

package main

import (
	"demo"
	"fmt"
)

func main() {
	context, _ := demo.InitialNatServer()
	_, err := context.Publish("demo.data", []byte("Hello"))
	if err != nil {
		fmt.Printf("Can not publish a new message to NAT server")
	}
	fmt.Printf("Done")
}

sub.go

package main

import (
	"demo"
	"fmt"
	"time"

	"github.com/nats-io/nats.go"
)

func main() {
	context, _ := demo.InitialNatServer()
	_, err := context.Subscribe("demo.data", receiveMessage)
	if err != nil {
		fmt.Printf("Can not subscribe from NAT server")
	}
	fmt.Println("Done")
	time.Sleep( 5 * time.Second)
}

func receiveMessage(m *nats.Msg) {
	err := m.Ack()
	if err != nil {
		fmt.Errorf("Can not send ack to NAT server")
	}
	fmt.Printf("Got data =%s\n", m.Data)
}