NATS

Intro

• Scalable and resilient message broker
• Written in Go programming language
  • Initial version written in Ruby (Derek Collison)
• Based on multiple cooperating components

NATS components

• Server (gnatsd)
• Interfaces for various ecosystems
• NATS Streaming Server
• NATS Connector Framework

NATS ecosystems support

Official

1. C
2. C#
3. Elixir
4. Go
5. Java
6. NGINX
7. Node.js
8. Python Asyncio
9. Python Tornado
10. Ruby
11. TypeScript

Unofficial

1. .NET
2. Arduino
3. Clojure
4. Elm
5. Erlang
6. Haskell
7. Lua
8. MicroPython
9. PHP
10. Perl
11. Python
12. Python Twisted
13. Qt5 C++
14. Rust
15. Scala
16. Spring API
17. Swift

Communication strategies

• pub-sub
  • using subjects (similar to topics)
• push-pull (via queues)
• request-reply
  • point-to-point
  • one-to-many
  • (based on inbox for replies)

Protocol used by NATS

• text-based
  • but data can be binary
• possible to test it via telnet

Command   send by
INFO      server
CONNECT   client
PUB       client
SUB       client
UNSUB     client
MSG       server
PING      client and server
PONG      client and server
+OK       server
-ERR      server

Communicating with NATS

• We’ll use Go programming language in all examples
• Possible to rewrite into other languages w/o problems

The simplest producer

• Without any error checks
• Not production-ready, of course

package main
 
import nats "github.com/nats-io/go-nats"
 
const Subject = "test1"
 
func main() {
        conn, _ := nats.Connect(nats.DefaultURL)
 
        conn.Publish(Subject, []byte("Hello World"))
 
        conn.Flush()
}

Better producer

• With error checks etc.

package main
 
import (
        nats "github.com/nats-io/go-nats"
        "log"
)
 
const Subject = "test1"
 
func main() {
        conn, err := nats.Connect(nats.DefaultURL)
 
        if err != nil {
                log.Fatal(err)
        }
 
        defer conn.Close()
 
        println("Connected")
 
        err2 := conn.Publish(Subject, []byte("Hello World"))
 
        if err2 != nil {
                log.Fatal(err2)
        }
 
        conn.Flush()
 
        println("Message sent")
}

Usage of channels

• This is the best solution in Go ecosystem

package main
 
import (
        nats "github.com/nats-io/go-nats"
        "log"
)
 
const Subject = "test1"
 
func main() {
        conn, err := nats.Connect(nats.DefaultURL)
 
        if err != nil {
                log.Fatal(err)
        }
 
        defer conn.Close()
 
        println("Connected")
 
        econn, err2 := nats.NewEncodedConn(conn, nats.DEFAULT_ENCODER)
 
        if err2 != nil {
                log.Fatal(err)
        }
 
        defer econn.Close()
 
        channel := make(chan string)
        econn.BindSendChan(Subject, channel)
 
        println("Channel created")
 
        channel <- "Hello World #1"
        channel <- "Hello World #2"
        channel <- "Hello World #3"
 
        println("All messages sent")
}

The simplest consumer

package main
 
import (
        "fmt"
        "sync"

        nats "github.com/nats-io/go-nats"
)
 
const Subject = "test1"
 
func main() {
        conn, _ := nats.Connect(nats.DefaultURL)
 
        wg := sync.WaitGroup{}
        wg.Add(1)
 
        conn.Subscribe(Subject, func(m *nats.Msg) {
                fmt.Printf("Received a message: %s\n", string(m.Data))
                wg.Done()
        })
        wg.Wait()
}

Better consumer

• Again with error checks etc.

package main
 
import (
        "fmt"
        nats "github.com/nats-io/go-nats"
        "log"
        "sync"
)
 
const Subject = "test1"
 
func main() {
        conn, err := nats.Connect(nats.DefaultURL)
 
        if err != nil {
                log.Fatal(err)
        }
 
        defer conn.Close()
 
        wg := sync.WaitGroup{}
        wg.Add(1)
 
        sub, err2 := conn.Subscribe(Subject, func(m *nats.Msg) {
                fmt.Printf("Received a message: %s\n", string(m.Data))
                wg.Done()
        })
 
        if err2 != nil {
                log.Fatal(err2)
        }
 
        println("Subscribed", sub)
 
        wg.Wait()
 
        println("Finished waiting for message")
 
        err3 := sub.Unsubscribe()
        if err3 != nil {
                log.Fatal(err3)
        }
 
        println("Unsubscribed")
}

Usage of channels in consumer code

package main
 
import (
        nats "github.com/nats-io/go-nats"
        "log"
)
 
const Subject = "test1"
 
func main() {
        conn, err := nats.Connect(nats.DefaultURL)
 
        if err != nil {
                log.Fatal(err)
        }
 
        defer conn.Close()
 
        econn, err2 := nats.NewEncodedConn(conn, nats.DEFAULT_ENCODER)
 
        if err2 != nil {
                log.Fatal(err)
        }
 
        defer econn.Close()
 
        channel := make(chan string)
        econn.BindRecvChan(Subject, channel)
 
        println("Channel created")
 
        println(<-channel)
        println(<-channel)
        println(<-channel)
}

Multi-channel consumer

• interesting solution for some subtle communication patterns

package main
 
import (
        nats "github.com/nats-io/go-nats"
        "log"
)
 
const Subject = "test1"
const Control = "test2"
 
func main() {
        conn, err := nats.Connect(nats.DefaultURL)
 
        if err != nil {
                log.Fatal(err)
        }
 
        defer conn.Close()
 
        econn, err2 := nats.NewEncodedConn(conn, nats.DEFAULT_ENCODER)
 
        if err2 != nil {
                log.Fatal(err2)
        }
 
        defer econn.Close()
 
        cconn, err3 := nats.NewEncodedConn(conn, nats.DEFAULT_ENCODER)
 
        if err3 != nil {
                log.Fatal(err3)
        }
 
        defer cconn.Close()
 
        data_channel := make(chan string)
        econn.BindRecvChan(Subject, data_channel)
 
        println("Data channel created")
 
        control_channel := make(chan string)
        cconn.BindRecvChan(Control, control_channel)
 
        println("Control channel created")
 
MESSAGE_LOOP:
        for {
                select {
                case message := <-data_channel:
                        println("Received data message", message)
                case control := <-control_channel:
                        println("Received control message", control)
                        if control == "EXIT" {
                                break MESSAGE_LOOP
                        }
                }
                println("--------")
        }
}

NATS Streaming Server

• Extension to the standard message broker
  • message persistence (optional)
  • ensure message delivery for producer (optional)
  • ensure message delivery for consumer (optional)
  • configurable publisher rate limit
  • configurable rate limiting per subscriber
  • messages are replayable
• NATS Streaming support
  1. C
  2. C#
  3. Go
  4. Java
  5. Node.js
  6. Python Asyncio
  7. Ruby

Links

Czech articles about NATS

• Komunikace s message brokery z programovacího jazyka Go
• Použití message brokeru NATS
• NATS Streaming Server