---
id: 8f79af5c6af89b748112aa8b77b348ab
author: prravda
title: Configuratie van de ingebouwde NATS-server in Go-applicaties en de structuur van inprocescommunicatie
description: Dit document behandelt de configuratie van een NATS-broker (nats-server/v2) die embedded binnen een Go-proces wordt gestart, alsmede het ontwerp van in-memory communicatie op basis van InProcessConn en een door de consument gedefinieerde interface.
language: nl
date: 2025-08-04T01:05:22.150350794Z
path: /blog/posts/how-embedded-nats-communicate-with-go-application-z36089af0
category: network
---

# Aan de slag

## Over NATS

> Softwareapplicaties en -diensten moeten gegevens uitwisselen. NATS is een infrastructuur die zulke gegevensuitwisseling mogelijk maakt, gesegmenteerd in de vorm van berichten. We noemen dit een "**message oriented middleware**".
>
> Met NATS kunnen applicatieontwikkelaars:
>
> - Moeiteloos gedistribueerde en schaalbare client-server-applicaties bouwen.
> - Gegevens in realtime op een algemene manier opslaan en distribueren. Dit kan flexibel worden bereikt in verschillende omgevingen, talen, cloudproviders en on-premises systemen.
>
> [Wat is NATS, NATS docs](https://docs.nats.io/nats-concepts/what-is-nats)

- NATS is een berichtenbroker geschreven in Go.

## Embedded NATS

> Als uw applicatie in Go is geschreven en als het past bij uw gebruiksscenario en implementatiescenario's, kunt u zelfs een NATS-server in uw applicatie insluiten.

[Embedding NATS, NATS docs](https://docs.nats.io/running-a-nats-service/clients#embedding-nats)

>

- Er is een bijzonderheid van NATS: voor applicaties die in Go zijn gebouwd, wordt de embedded mode ondersteund.
- Dat wil zeggen dat de broker zelf in een applicatie die in Go is gemaakt, kan worden ingebed (embed), in plaats van de gebruikelijke werkwijze van berichtenbrokers, waarbij een afzonderlijke brokerserver wordt gestart en de applicatie via een client met die server communiceert.

## Voordelen en gebruiksscenario's van embedded NATS

- Er is een goed uitgelegde [Youtube-video](https://youtu.be/cdTrl8UfcBo?si=KYcXOQpiyLUft6AN), dus ik verwijs naar die video in plaats van het hier uitgebreid te behandelen.
- Zonder een afzonderlijke berichtenbrokerserver te implementeren, kan men een modular monolith applicatie bouwen en zo separation of concerns realiseren, en profiteren van het voordeel dat NATS als embedded component kan worden opgenomen. Bovendien wordt implementatie als single binary mogelijk.
- Naast platforms zonder netwerk (wasm) kan het ook nuttig zijn in offline-first applicaties.

# Voorbeeld uit de officiële documentatie

```go
package main

import (
    "fmt"
    "time"

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

func main() {
    opts := &server.Options{}

    // Initialize new server with options
    ns, err := server.NewServer(opts)

    if err != nil {
        panic(err)
    }

    // Start the server via goroutine
    go ns.Start()

    // Wait for server to be ready for connections
    if !ns.ReadyForConnections(4 * time.Second) {
        panic("not ready for connection")
    }

    // Connect to server
    nc, err := nats.Connect(ns.ClientURL())

    if err != nil {
        panic(err)
    }

    subject := "my-subject"

    // Subscribe to the subject
    nc.Subscribe(subject, func(msg *nats.Msg) {
        // Print message data
        data := string(msg.Data)
        fmt.Println(data)

        // Shutdown the server (optional)
        ns.Shutdown()
    })

    // Publish data to the subject
    nc.Publish(subject, []byte("Hello embedded NATS!"))

    // Wait for server shutdown
    ns.WaitForShutdown()
}
```

- Dit is het [voorbeeld van Embedded NATS](https://dev.to/karanpratapsingh/embedding-nats-in-go-19o) dat de officiële NATS-documentatie aanhaalt, maar als je de code uit dit voorbeeld volgt, vindt de communicatie niet plaats in embedding mode.

```bash
Every 2.0s: netstat -an | grep 127.0.0.1         pravdalaptop-home.local: 02:34:20
                                                                     in 0.017s (0)
...
tcp4       0      0  127.0.0.1.4222         127.0.0.1.63769        TIME_WAIT
```

- Wanneer je het netwerkverkeer naar [localhost](http://localhost) (127.0.0.1) controleert met het commando `watch 'netstat -an | grep 127.0.0.1'` en het Go-bestand uitvoert met `go run .`, zie je nieuwe netwerkverzoeken ontstaan vanaf de standaardpoort van NATS, `4222`.

# Juiste configuraties voor embedding mode

- Om de communicatie in embedded mode te laten verlopen zoals bedoeld, zijn de volgende twee opties nodig.
  - Client: de optie `InProcessServer` moet worden meegegeven.
  - Server: in `Server.Options` moet de flag `DontListen` expliciet op `true` worden gezet.
- Deze onderdelen zijn niet officieel gedocumenteerd; het begin van deze functionaliteit kan worden teruggevonden in deze [PR](https://github.com/nats-io/nats-server/pull/2360).

  > This PR adds three things:
  >
  > 1. `InProcessConn()` function to `Server` which builds a `net.Pipe` to get a connection to the NATS server without using TCP sockets
  > 2. `DontListen` option which tells the NATS server not to listen on the usual TCP listener
  > 3. `startupComplete` channel, which is closed right before we start `AcceptLoop`, and `readyForConnections` will wait for it
  >
  > The main motivation for this is that we have an application that can run either in a monolith (single-process) mode or a polylith (multi-process) mode. We'd like to be able to use NATS for both modes for simplicity, but the monolith mode has to be able to cater for a variety of platforms where opening socket connections either doesn't make sense (mobile) or just isn't possible (WASM). These changes will allow us to use NATS entirely in-process instead.
  >
  > An accompanying PR [nats-io/nats.go#774](https://github.com/nats-io/nats.go/pull/774) adds support to the client side.
  >
  > This is my first PR to this project so apologies in advance if I've missed anything obvious anywhere.
  >
  > /cc @nats-io/core

# Werkend voorbeeld voor embedded mode

```go
package main

import (
	"fmt"
	"time"

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

func main() {
	opts := &server.Options{
		// for configuring the embeded NATS server
		// set DonListen as true
		DontListen: true,
	}

	// Initialize new server with options
	ns, err := server.NewServer(opts)

	if err != nil {
		panic(err)
	}

	// Start the server via goroutine
	go ns.Start()

	// Wait for server to be ready for connections
	if !ns.ReadyForConnections(10 * time.Second) {
		panic("not ready for connection")
	}

	// Connect to server via in-process connection
	nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))

	if err != nil {
		panic(err)
	}

	subject := "my-subject"

	// Subscribe to the subject
	nc.Subscribe(subject, func(msg *nats.Msg) {
		// Print message data
		data := string(msg.Data)
		fmt.Println(data)

		// Shutdown the server (optional)
		ns.Shutdown()
	})

	// Publish data to the subject
	nc.Publish(subject, []byte("Hello embedded NATS!"))

	// Wait for server shutdown
	ns.WaitForShutdown()
}

```

```bash
Every 2.0s: netstat -an | grep 127.0.0.1         pravdalaptop-home.local: 02:37:50
                                                                     in 0.023s (0)
...no additional logs

```

- Nu zie je dat er zoals bedoeld geen extra network hop meer optreedt.

# Onder de motorkap

## TL;DR

![diagram1](/assets/images/embedded-nats/seq1.svg)

- Dit is een sequence diagram dat laat zien welke functies er intern worden aangeroepen en hoe ze werken wanneer deze code in `main.go` wordt uitgevoerd. De kern kan als volgt worden samengevat.
  - Via `DontListen: true` slaat de server de client listening phase over die `AcceptLoop` heet.
  - Wanneer de Connect-optie `InProcessServer` van de client is ingeschakeld, wordt een in-memory connection gemaakt; via `net.Pipe` wordt een pipe aangemaakt en wordt het andere uiteinde van de pipe als `net.Conn`-type aan de client teruggegeven.
  - Client en server voeren de in-process communicatie uit via die connection.

## Server

### AcceptLoop

```go
// nats-server/server/server.go

// Wait for clients.
if !opts.DontListen {
	s.AcceptLoop(clientListenReady)
}
```

- Wanneer `DontListen` true is, wordt eerst de client listening phase die `AcceptLoop` heet overgeslagen.

```go
// nats-server/server/server.go

// AcceptLoop is exported for easier testing.
func (s *Server) AcceptLoop(clr chan struct{}) {
	// If we were to exit before the listener is setup properly,
	// make sure we close the channel.
	defer func() {
		if clr != nil {
			close(clr)
		}
	}()

	if s.isShuttingDown() {
		return
	}

	// Snapshot server options.
	opts := s.getOpts()

	// Setup state that can enable shutdown
	s.mu.Lock()
	hp := net.JoinHostPort(opts.Host, strconv.Itoa(opts.Port))
	l, e := natsListen("tcp", hp)
	s.listenerErr = e
	if e != nil {
		s.mu.Unlock()
		s.Fatalf("Error listening on port: %s, %q", hp, e)
		return
	}
	s.Noticef("Listening for client connections on %s",
		net.JoinHostPort(opts.Host, strconv.Itoa(l.Addr().(*net.TCPAddr).Port)))

	// Alert of TLS enabled.
	if opts.TLSConfig != nil {
		s.Noticef("TLS required for client connections")
		if opts.TLSHandshakeFirst && opts.TLSHandshakeFirstFallback == 0 {
			s.Warnf("Clients that are not using \"TLS Handshake First\" option will fail to connect")
		}
	}

	// If server was started with RANDOM_PORT (-1), opts.Port would be equal
	// to 0 at the beginning this function. So we need to get the actual port
	if opts.Port == 0 {
		// Write resolved port back to options.
		opts.Port = l.Addr().(*net.TCPAddr).Port
	}

	// Now that port has been set (if it was set to RANDOM), set the
	// server's info Host/Port with either values from Options or
	// ClientAdvertise.
	if err := s.setInfoHostPort(); err != nil {
		s.Fatalf("Error setting server INFO with ClientAdvertise value of %s, err=%v", opts.ClientAdvertise, err)
		l.Close()
		s.mu.Unlock()
		return
	}
	// Keep track of client connect URLs. We may need them later.
	s.clientConnectURLs = s.getClientConnectURLs()
	s.listener = l

	go s.acceptConnections(l, "Client", func(conn net.Conn) { s.createClient(conn) },
		func(_ error) bool {
			if s.isLameDuckMode() {
				// Signal that we are not accepting new clients
				s.ldmCh <- true
				// Now wait for the Shutdown...
				<-s.quitCh
				return true
			}
			return false
		})
	s.mu.Unlock()

	// Let the caller know that we are ready
	close(clr)
	clr = nil
}
```

- Ter info: de AcceptLoop-functie doorloopt de volgende stappen. Het gaat om onderdelen die met netwerkcommunicatie te maken hebben, zoals `TLS` en `hostPort`. Bij in-process communicatie zijn deze niet nodig, dus ze mogen worden overgeslagen.

## Client

### InProcessServer

```go

// nats-go/nats.go

// Connect will attempt to connect to the NATS system.
// The url can contain username/password semantics. e.g. nats://derek:pass@localhost:4222
// Comma separated arrays are also supported, e.g. urlA, urlB.
// Options start with the defaults but can be overridden.
// To connect to a NATS Server's websocket port, use the `ws` or `wss` scheme, such as
// `ws://localhost:8080`. Note that websocket schemes cannot be mixed with others (nats/tls).
func Connect(url string, options ...Option) (*Conn, error) {
	opts := GetDefaultOptions()
	opts.Servers = processUrlString(url)
	for _, opt := range options {
		if opt != nil {
			if err := opt(&opts); err != nil {
				return nil, err
			}
		}
	}
	return opts.Connect()
}
```

```go
// nats-go/nats.go

// Options can be used to create a customized connection.
type Options struct {
	// Url represents a single NATS server url to which the client
	// will be connecting. If the Servers option is also set, it
	// then becomes the first server in the Servers array.
	Url string

	// InProcessServer represents a NATS server running within the
	// same process. If this is set then we will attempt to connect
	// to the server directly rather than using external TCP conns.
	InProcessServer InProcessConnProvider

	//...
}
```

```go
// nats-go/nats.go

type InProcessConnProvider interface {
	InProcessConn() (net.Conn, error)
}
```

- De `Connect`-functie, die de verbinding tussen de NATS-server en de NATS-client tot stand brengt, accepteert een client URL en connect-opties. De struct Options, waarin deze opties worden verzameld, bevat een field `InProcessServer` van het interfacetype `InProcessConnProvider`.

```go
// main.go of example code

// Initialize new server with options
ns, err := server.NewServer(opts)

//...

// Connect to server via in-process connection
nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))
```

- Wanneer de NATS-client verbinding maakt via `Connect` en `nats.InProcessServer(ns)` wordt doorgegeven,

```go
// nats-go/nats.go

// InProcessServer is an Option that will try to establish a direction to a NATS server
// running within the process instead of dialing via TCP.
func InProcessServer(server InProcessConnProvider) Option {
	return func(o *Options) error {
		o.InProcessServer = server
		return nil
	}
}
```

- wordt de InProcessServer-optie vervangen door de embedded NATS-server,

```go
// nats-go/nats.go

// createConn will connect to the server and wrap the appropriate
// bufio structures. It will do the right thing when an existing
// connection is in place.
func (nc *Conn) createConn() (err error) {
	if nc.Opts.Timeout < 0 {
		return ErrBadTimeout
	}
	if _, cur := nc.currentServer(); cur == nil {
		return ErrNoServers
	}

	// If we have a reference to an in-process server then establish a
	// connection using that.
	if nc.Opts.InProcessServer != nil {
		conn, err := nc.Opts.InProcessServer.InProcessConn()
		if err != nil {
			return fmt.Errorf("failed to get in-process connection: %w", err)
		}
		nc.conn = conn
		nc.bindToNewConn()
		return nil
	}

	//...
}
```

- Dit interface wordt gebruikt in de functie `createConn` die de connection tot stand brengt. Wanneer de optie `InProcessServer` niet nil is (dus geldig), wordt de `InProcessConn` van de InProcessServer-optie uitgevoerd,

```go
// nats-server/server/server.go

// InProcessConn returns an in-process connection to the server,
// avoiding the need to use a TCP listener for local connectivity
// within the same process. This can be used regardless of the
// state of the DontListen option.
func (s *Server) InProcessConn() (net.Conn, error) {
	pl, pr := net.Pipe()
	if !s.startGoRoutine(func() {
		s.createClientInProcess(pl)
		s.grWG.Done()
	}) {
		pl.Close()
		pr.Close()
		return nil, fmt.Errorf("failed to create connection")
	}
	return pr, nil
}
```

- De `InProcessConn` die in de server is geïmplementeerd, wordt aangeroepen.
- Deze functie wordt aangeroepen vanuit `nats.go`, de Go-client van NATS, wanneer de `InProcessServer` van de nc (NATS connection) niet nil is. Ze maakt een connection (`net.Conn`) aan en bindt die aan de connection van de server.

# Consumer driven interface van Go

> A type implements an interface by implementing its methods. There is no explicit declaration of intent, no "implements" keyword. Implicit interfaces decouple the definition of an interface from its implementation, which could then appear in any package without prearrangement.

[Interfaces are implemented implicitly, A Tour of Go](https://go.dev/tour/methods/10)

> If a type exists only to implement an interface and will never have exported methods beyond that interface, there is no need to export the type itself.

[Generality, Effective Go](https://go.dev/doc/effective_go#generality)

- Dit interfaceontwerp belicht goed wat in Go vaak consumer defined interface en structural typing (duck typing) worden genoemd. Daarom wil ik dit onderwerp ook kort introduceren.

```go
// nats-go/nats.go

// Options can be used to create a customized connection.
type Options struct {
	// Url represents a single NATS server url to which the client
	// will be connecting. If the Servers option is also set, it
	// then becomes the first server in the Servers array.
	Url string

	// InProcessServer represents a NATS server running within the
	// same process. If this is set then we will attempt to connect
	// to the server directly rather than using external TCP conns.
	InProcessServer InProcessConnProvider

	//...
}
```

```go
// nats-go/nats.go

type InProcessConnProvider interface {
	InProcessConn() (net.Conn, error)
}
```

- Terug naar de code. In de nats.go-client is het field `InProcessServer` van de optiestruct gedefinieerd als het interface `InProcessConnProvider`, dat alleen `InProcessConn` uitvoert.

```go
// nats-server/server/server.go

// InProcessConn returns an in-process connection to the server,
// avoiding the need to use a TCP listener for local connectivity
// within the same process. This can be used regardless of the
// state of the DontListen option.
func (s *Server) InProcessConn() (net.Conn, error) {
	pl, pr := net.Pipe()
	if !s.startGoRoutine(func() {
		s.createClientInProcess(pl)
		s.grWG.Done()
	}) {
		pl.Close()
		pr.Close()
		return nil, fmt.Errorf("failed to create connection")
	}
	return pr, nil
}
```

- Het type dat daarin wordt gestopt, is echter de `Server` uit nats-server, die naast `InProcessConn` ook allerlei andere functionaliteit heeft.
- De reden is dat de client in deze situatie alleen belang hecht aan de vraag of het interface `InProcessConn` is aangeboden; de overige functionaliteit is voor de client niet van groot belang.
- Daarom gebruikt de nats.go-client alleen het consumer defined interface `InProcessConnProvider`, dat uitsluitend de functionaliteit `InProcessConn() (net.Conn, error)` definieert.

# Conclusie

- Kort besproken zijn de embedded mode van NATS, de werking ervan, en het consumer defined interface van Go dat in de code van NATS zichtbaar is.
- Ik hoop dat deze informatie nuttig is voor mensen die NATS met een soortgelijk doel gebruiken, en hiermee sluit ik dit artikel af.