Embedding NATS server in a GO application

 

In this post we examine two methods of embedding a NATS server as part of a GO application. This post is base on this video.

First we we use a NATS server that includes a listener so we can have NATS connections from the external world, and then we run NATS server without a listener.

Embedded NATS server with a Listener

This can be used when we want the application to run NATS server embedded within the GO process, and we still want to preserve the ability to get messages from external clients through the network. We can even run this NATS servers as part of a cluster.

package main

import (
  "errors"
  "fmt"
  "github.com/nats-io/nats-server/v2/server"
  "github.com/nats-io/nats.go"
  "time"
)

func main() {
  options := server.Options{}
  natsServer, err := server.NewServer(&options)
  if err != nil {
   panic(err)
  }

  natsServer.ConfigureLogger()

  go natsServer.Start()

  if !natsServer.ReadyForConnections(time.Minute) {
   panic(errors.New("nats server not ready"))
  }

  natsConnection, err := nats.Connect(natsServer.ClientURL())
  if err != nil {
   panic(err)
  }

  natsSubscription, err := natsConnection.Subscribe("queue1", func(message *nats.Msg) {
   response := fmt.Sprintf("got your message: %s", string(message.Data))
   err := message.Respond([]byte(response))
   if err != nil {
    panic(err)
   }
  })

  if err != nil {
   panic(err)
  }

  for i := range 10 {
   message := fmt.Sprintf("My %v message", i)
   response, err := natsConnection.Request("queue1", []byte(message), time.Second)
   if err != nil {
    panic(err)
   }
   fmt.Printf("got response: %v\n", string(response.Data))
  }

  err = natsSubscription.Unsubscribe()
  if err != nil {
   panic(err)
  }

  natsServer.WaitForShutdown()
}

And the output is:

[19234] [INF] Starting nats-server
[19234] [INF]   Version:  2.10.17
[19234] [INF]   Git:      [not set]
[19234] [INF]   Name:     NCD4DRH4CVNVY44VGEPOTKKJMGM524KCUN26GQNQTXYYKCGNEXUNIZ6P
[19234] [INF]   ID:       NCD4DRH4CVNVY44VGEPOTKKJMGM524KCUN26GQNQTXYYKCGNEXUNIZ6P
[19234] [INF] Listening for client connections on 0.0.0.0:4222
[19234] [INF] Server is ready
got response: got your message: My 0 message
got response: got your message: My 1 message
got response: got your message: My 2 message
got response: got your message: My 3 message
got response: got your message: My 4 message
got response: got your message: My 5 message
got response: got your message: My 6 message
got response: got your message: My 7 message
got response: got your message: My 8 message
got response: got your message: My 9 message
^C[19234] [INF] Initiating Shutdown...
[19234] [INF] Server Exiting..

Embedded NATS server with NO Listener

This can be used for an application that requires NATS only for its internal message distribution, and separation between internal modules. Using in-process API instead of localhost communication is much faster.

To do this, we configure the NATS server to "DontListen", and use a in-process option for the client connection.

package main

import (
  "errors"
  "fmt"
  "github.com/nats-io/nats-server/v2/server"
  "github.com/nats-io/nats.go"
  "time"
)

func main() {
  serverOptions := server.Options{
   DontListen: true,
  }
  natsServer, err := server.NewServer(&serverOptions)
  if err != nil {
   panic(err)
  }

  natsServer.ConfigureLogger()

  go natsServer.Start()

  if !natsServer.ReadyForConnections(time.Minute) {
   panic(errors.New("nats server not ready"))
  }

  option := nats.InProcessServer(natsServer)
  natsConnection, err := nats.Connect(natsServer.ClientURL(), option)
  if err != nil {
   panic(err)
  }

  natsSubscription, err := natsConnection.Subscribe("queue1", func(message *nats.Msg) {
   response := fmt.Sprintf("got your message: %s", string(message.Data))
   err := message.Respond([]byte(response))
   if err != nil {
    panic(err)
   }
  })

  if err != nil {
   panic(err)
  }

  for i := range 10 {
   message := fmt.Sprintf("My %v message", i)
   response, err := natsConnection.Request("queue1", []byte(message), time.Second)
   if err != nil {
    panic(err)
   }
   fmt.Printf("got response: %v\n", string(response.Data))
  }

  err = natsSubscription.Unsubscribe()
  if err != nil {
   panic(err)
  }

  natsServer.WaitForShutdown()
}

and the output is:

[20802] [INF] Starting nats-server
[20802] [INF]   Version:  2.10.17
[20802] [INF]   Git:      [not set]
[20802] [INF]   Name:     NB2IB6QLEWN2XNPJUAVIWKRICFJUIQJ2CRQWGOC2XKV7IFO36HOK7WXM
[20802] [INF]   ID:       NB2IB6QLEWN2XNPJUAVIWKRICFJUIQJ2CRQWGOC2XKV7IFO36HOK7WXM
[20802] [INF] Server is ready
got response: got your message: My 0 message
got response: got your message: My 1 message
got response: got your message: My 2 message
got response: got your message: My 3 message
got response: got your message: My 4 message
got response: got your message: My 5 message
got response: got your message: My 6 message
got response: got your message: My 7 message
got response: got your message: My 8 message
got response: got your message: My 9 message
^C[20802] [INF] Initiating Shutdown...
[20802] [INF] Server Exiting..