---
id: 8f79af5c6af89b748112aa8b77b348ab
author: prravda
title: Configurazione di un server NATS integrato in un'applicazione Go e struttura della comunicazione in-process
description: Vengono trattati la configurazione per l'avvio di un broker NATS (nats-server/v2) incorporato all'interno di un processo Go, nonché la comunicazione in memoria basata su InProcessConn e la progettazione di interfacce definite dal consumatore.
language: it
date: 2025-08-04T01:05:22.150350794Z
path: /blog/posts/how-embedded-nats-communicate-with-go-application-z36089af0
category: network
---

# Per iniziare

## Informazioni su NATS

> Le applicazioni software e i servizi devono scambiarsi dati. NATS è un'infrastruttura che consente tale scambio di dati, suddivisi sotto forma di messaggi. La definiamo "**message oriented middleware**".
>
> Con NATS, gli sviluppatori di applicazioni possono:
>
> - Costruire in modo semplice applicazioni client-server distribuite e scalabili.
> - Memorizzare e distribuire dati in tempo reale in modo generale. Ciò può essere realizzato in modo flessibile across vari ambienti, linguaggi, provider cloud e sistemi on-premises.
>
> [What is NATS, NATS docs](https://docs.nats.io/nats-concepts/what-is-nats)

- NATS è un message broker costruito in Go.

## NATS incorporato

> Se la tua applicazione è scritta in Go e se si adatta al tuo caso d'uso e agli scenari di deployment, puoi persino incorporare un server NATS all'interno della tua applicazione.

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

>

- Una particolarità di NATS è che, per le applicazioni costruite in Go, supporta la modalità embedded.
- In altre parole, anziché il metodo tipico dei message broker, ovvero avviare un server broker separato e comunicare con esso tramite i client dell'applicazione, è possibile incorporare il broker stesso (embed) nell'applicazione realizzata in Go.

## Vantaggi e casi d'uso di NATS incorporato

- Poiché esiste un [video su YouTube](https://youtu.be/cdTrl8UfcBo?si=KYcXOQpiyLUft6AN) ben curato, si rimanda al link del video.
- Anche senza distribuire un server di message broker separato, è possibile creare un'applicazione modular monolith e ottenere la separation of concerns, traendo vantaggio dalla possibilità di incorporare NATS in modalità embedded. Inoltre, diventa possibile il single binary deployment.
- Può essere utile non solo per la platform with no network (wasm), ma anche per le offline-first application.

# Esempio nella documentazione ufficiale

```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{}

    // Inizializza un nuovo server con le opzioni
    ns, err := server.NewServer(opts)

    if err != nil {
        panic(err)
    }

    // Avvia il server tramite goroutine
    go ns.Start()

    // Attende che il server sia pronto per le connessioni
    if !ns.ReadyForConnections(4 * time.Second) {
        panic("not ready for connection")
    }

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

    if err != nil {
        panic(err)
    }

    subject := "my-subject"

    // Sottoscrizione al subject
    nc.Subscribe(subject, func(msg *nats.Msg) {
        // Stampa i dati del messaggio
        data := string(msg.Data)
        fmt.Println(data)

        // Arresta il server (opzionale)
        ns.Shutdown()
    })

    // Pubblica dati sul subject
    nc.Publish(subject, []byte("Hello embedded NATS!"))

    // Attende l'arresto del server
    ns.WaitForShutdown()
}
```

- Si tratta dell'[esempio di NATS incorporato](https://dev.to/karanpratapsingh/embedding-nats-in-go-19o) riportato dalla documentazione ufficiale di NATS, ma seguendo il codice di esempio la comunicazione non avviene in modalità embedding.

```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
```

- Eseguendo `watch 'netstat -an | grep 127.0.0.1'` per monitorare la rete che transita su [localhost](http://localhost) (127.0.0.1) e avviando il file go con `go run .`, si può osservare l'aggiunta di nuove richieste di rete che partono dalla porta predefinita di NATS, `4222`.

# Configurazioni corrette per la modalità embedding

- Per comunicare nella modalità embedded come previsto sono necessarie le seguenti due opzioni.
  - Client: è necessario inserire l'opzione `InProcessServer`.
  - Server: è necessario impostare esplicitamente il flag `DontListen` di `Server.Options` a `true`.
- Tali aspetti non sono ufficialmente documentati; l'origine di questa funzionalità è rintracciabile nella seguente [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

# Esempio funzionante per la modalità embedded

```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{
		// per configurare il server NATS incorporato
		// impostare DonListen a true
		DontListen: true,
	}

	// Inizializza un nuovo server con le opzioni
	ns, err := server.NewServer(opts)

	if err != nil {
		panic(err)
	}

	// Avvia il server tramite goroutine
	go ns.Start()

	// Attende che il server sia pronto per le connessioni
	if !ns.ReadyForConnections(10 * time.Second) {
		panic("not ready for connection")
	}

	// Connessione al server tramite connessione in-process
	nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))

	if err != nil {
		panic(err)
	}

	subject := "my-subject"

	// Sottoscrizione al subject
	nc.Subscribe(subject, func(msg *nats.Msg) {
		// Stampa i dati del messaggio
		data := string(msg.Data)
		fmt.Println(data)

		// Arresta il server (opzionale)
		ns.Shutdown()
	})

	// Pubblica dati sul subject
	nc.Publish(subject, []byte("Hello embedded NATS!"))

	// Attende l'arresto del server
	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

```

- Ora si può osservare che non si verifica alcun network hop aggiuntivo, come previsto.

# Sotto il cofano

## TL;DR

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

- Si tratta di un sequence diagram che mostra quali funzioni vengono eseguite e in che modo quando il codice viene eseguito da `main.go`; i punti principali sono i seguenti.
  - Tramite `DontListen: true`, il server omette la fase di ascolto dei client denominata `AcceptLoop`.
  - Se l'opzione `InProcessServer` del Connect del client è attiva, viene creata una connessione in-memory tramite `net.Pipe`, e l'estremità della pipe viene restituita al client come tipo `net.Conn`.
  - Client e server comunicano in-process attraverso tale connessione.

## Server

### AcceptLoop

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

// Attende i client.
if !opts.DontListen {
	s.AcceptLoop(clientListenReady)
}
```

- Innanzitutto, quando `DontListen` è true, la fase di ascolto dei client denominata `AcceptLoop` viene omessa.

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

// AcceptLoop è esportata per facilitare i test.
func (s *Server) AcceptLoop(clr chan struct{}) {
	// Se uscissimo prima che il listener sia configurato correttamente,
	// assicuriamoci di chiudere il canale.
	defer func() {
		if clr != nil {
			close(clr)
		}
	}()

	if s.isShuttingDown() {
		return
	}

	// Cattura uno snapshot delle opzioni del server.
	opts := s.getOpts()

	// Configura lo stato che consente lo 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)))

	// Segnala che TLS è abilitato.
	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")
		}
	}

	// Se il server è stato avviato con RANDOM_PORT (-1), opts.Port sarebbe uguale
	// a 0 all'inizio di questa funzione. Quindi dobbiamo ottenere la porta effettiva
	if opts.Port == 0 {
		// Riscrive la porta risolta nelle opzioni.
		opts.Port = l.Addr().(*net.TCPAddr).Port
	}

	// Ora che la porta è stata impostata (se era RANDOM), imposta
	// Host/Port delle INFO del server con i valori di Options o
	// 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
	}
	// Tiene traccia degli URL di connessione dei client. Potrebbero servire più avanti.
	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() {
				// Segnala che non stiamo accettando nuovi client
				s.ldmCh <- true
				// Ora attende lo Shutdown...
				<-s.quitCh
				return true
			}
			return false
		})
	s.mu.Unlock()

	// Comunica al chiamante che siamo pronti
	close(clr)
	clr = nil
}
```

- Va notato che la funzione AcceptLoop esegue i passaggi seguenti. Le parti relative alla comunicazione di rete, come `TLS` o `hostPort`, non sono necessarie per la comunicazione in-process e possono quindi essere omesse.

## Client

### InProcessServer

```go

// nats-go/nats.go

// Connect tenta di connettersi al sistema NATS.
// L'url può contenere la semantica username/password. es. nats://derek:pass@localhost:4222
// Sono supportati anche array separati da virgole, es. urlA, urlB.
// Le opzioni partono dai valori predefiniti ma possono essere sovrascritte.
// Per connettersi alla porta websocket di un server NATS, usare lo schema `ws` o `wss`, come
// `ws://localhost:8080`. Si noti che gli schemi websocket non possono essere combinati con altri (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 può essere usato per creare una connessione personalizzata.
type Options struct {
	// Url rappresenta un singolo url del server NATS a cui il client
	// si connetterà. Se è impostata anche l'opzione Servers, diventa
	// il primo server nell'array Servers.
	Url string

	// InProcessServer rappresenta un server NATS in esecuzione all'interno
	// dello stesso processo. Se è impostato, tenteremo di connetterci
	// direttamente al server anziché usare connessioni TCP esterne.
	InProcessServer InProcessConnProvider

	//...
}
```

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

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

- La funzione `Connect`, che realizza la connessione tra il server NATS e il client NATS, accetta l'URL del client e le opzioni di connessione; la struct Options, che raccoglie tali opzioni, contiene il campo `InProcessServer` di tipo interface `InProcessConnProvider`.

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

// Inizializza un nuovo server con le opzioni
ns, err := server.NewServer(opts)

//...

// Connessione al server tramite connessione in-process
nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))
```

- Quando il client NATS esegue Connect e passa `nats.InProcessServer(ns)` nel campo `InProcessServer`,

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

// InProcessServer è un'Option che tenta di stabilire una connessione verso un server NATS
// in esecuzione all'interno del processo, anziché effettuare il dial tramite TCP.
func InProcessServer(server InProcessConnProvider) Option {
	return func(o *Options) error {
		o.InProcessServer = server
		return nil
	}
}
```

- l'InProcessServer dell'opzione viene sostituito con il server NATS incorporato,

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

// createConn si connette al server e avvolge le
// strutture bufio appropriate. Gestisce correttamente il caso in cui
// sia già presente una connessione esistente.
func (nc *Conn) createConn() (err error) {
	if nc.Opts.Timeout < 0 {
		return ErrBadTimeout
	}
	if _, cur := nc.currentServer(); cur == nil {
		return ErrNoServers
	}

	// Se abbiamo un riferimento a un server in-process, stabiliamo una
	// connessione usando quello.
	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
	}

	//...
}
```

- Questa interface viene utilizzata nella funzione `createConn`, che crea la connessione: se l'opzione `InProcessServer` non è nil (quindi valida), viene eseguito il metodo `InProcessConn` dell'InProcessServer contenuto nell'opzione,

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

// InProcessConn restituisce una connessione in-process al server,
// evitando la necessità di usare un listener TCP per la connettività locale
// all'interno dello stesso processo. Può essere usato indipendentemente
// dallo stato dell'opzione DontListen.
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
}
```

- Viene quindi invocato il `InProcessConn` implementato nel server.
- Questa funzione viene chiamata da `nats.go`, il client Go di NATS, quando l'`InProcessServer` dell'nc (connessione NATS) non è nil: crea la connessione (`net.Conn`) e la associa alla connessione del server.

# Interface Go definita dal consumer

> 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)

- Questo design delle interface rispecchia bene le cosiddette consumer defined interface e lo structural typing (duck typing) tipici di Go, per cui si è deciso di presentare anche questo argomento.

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

// Options può essere usato per creare una connessione personalizzata.
type Options struct {
	// Url rappresenta un singolo url del server NATS a cui il client
	// si connetterà. Se è impostata anche l'opzione Servers, diventa
	// il primo server nell'array Servers.
	Url string

	// InProcessServer rappresenta un server NATS in esecuzione all'interno
	// dello stesso processo. Se è impostato, tenteremo di connetterci
	// direttamente al server anziché usare connessioni TCP esterne.
	InProcessServer InProcessConnProvider

	//...
}
```

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

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

- Torniamo al codice. Nel client nats.go, il campo dell'opzione `InProcessServer` è definito come interface `InProcessConnProvider`, che prevede solo `InProcessConn`.

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

// InProcessConn restituisce una connessione in-process al server,
// evitando la necessità di usare un listener TCP per la connettività locale
// all'interno dello stesso processo. Può essere usato indipendentemente
// dallo stato dell'opzione DontListen.
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
}
```

- Tuttavia, il tipo che viene inserito è il `Server` di nats-server, che svolge molte altre funzioni oltre a `InProcessConn`.
- Questo perché, in questo contesto, l'interesse del client riguarda solo se è stata fornita l'interface `InProcessConn` o meno, mentre il resto non è rilevante.
- Pertanto, il client nats.go utilizza solo l'interface `InProcessConnProvider`, un'interface definita dal consumer che prevede unicamente la funzionalità `InProcessConn() (net.Conn, error)`.

# Conclusione

- Abbiamo trattato in sintesi la modalità embedded di NATS, il suo funzionamento e la consumer defined interface di Go che si può osservare nel codice di NATS.
- Auspico che queste informazioni siano utili a chi utilizza NATS con questi scopi, e con questo concludo l'articolo.