---
id: 8f79af5c6af89b748112aa8b77b348ab
author: prravda
title: Konfigurasi server NATS tertanam dalam aplikasi Go dan arsitektur komunikasi in-process
description: Dokumen ini membahas cara mengonfigurasi dan menjalankan broker NATS (nats-server/v2) yang tertanam di dalam proses Go, serta desain antarmuka konsumen dan komunikasi dalam memori berbasis InProcessConn.
language: id
date: 2025-08-04T01:05:22.150350794Z
path: /blog/posts/how-embedded-nats-communicate-with-go-application-z36089af0
category: network
---

# Memulai

## Tentang NATS

> Aplikasi dan layanan perangkat lunak perlu bertukar data. NATS adalah infrastruktur yang memungkinkan pertukaran data tersebut, yang tersegmentasi dalam bentuk pesan. Kami menyebut ini sebagai "**message oriented middleware**".
>
> Dengan NATS, pengembang aplikasi dapat:
>
> - Membangun aplikasi client-server yang terdistribusi dan dapat diskalakan dengan mudah.
> - Menyimpan dan mendistribusikan data secara real-time dengan cara umum. Hal ini dapat dicapai secara fleksibel di berbagai lingkungan, bahasa, penyedia cloud, dan sistem on-premises.
>
> [What is NATS, NATS docs](https://docs.nats.io/nats-concepts/what-is-nats)

- NATS merupakan message broker yang dibangun dengan Go.

## Embedded NATS

> Jika aplikasi Anda menggunakan Go, dan jika sesuai dengan kasus penggunaan serta skenario deployment Anda, Anda bahkan dapat menyematkan server NATS di dalam aplikasi Anda.

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

>

- Dan ada satu keunikan NATS, yaitu untuk aplikasi yang dibangun dengan Go, NATS mendukung embedded mode.
- Artinya, alih-alih struktur umum message broker, yaitu menjalankan server broker terpisah lalu berkomunikasi dengan server tersebut melalui client aplikasi, broker itu sendiri dapat di-embed (disematkan) ke dalam aplikasi yang dibuat dengan Go.

## Manfaat dan kasus penggunaan embedded NATS

- Terdapat [video Youtube](https://youtu.be/cdTrl8UfcBo?si=KYcXOQpiyLUft6AN) yang dijelaskan dengan baik, sehingga sebagai gantinya tautan video tersebut disertakan di sini.
- Tanpa perlu men-deploy server message broker terpisah, kita dapat membuat modular monolith application dan mencapai separation of concerns, sekaligus memanfaatkan keunggulan dapat menanamkan NATS sebagai embedded. Selain itu, single binary deployment juga menjadi mungkin.
- Selain untuk platform tanpa jaringan (wasm), NATS juga berguna pada offline-first application.

# Contoh pada dokumentasi resmi

```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()
}
```

- Ini adalah [contoh Embedded NATS](https://dev.to/karanpratapsingh/embedding-nats-in-go-19o) yang tercantum di dokumentasi resmi NATS, namun jika kode contoh tersebut diikuti, komunikasi tidak akan terjadi dalam 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
```

- Dengan menjalankan perintah `watch 'netstat -an | grep 127.0.0.1'` untuk memantau jaringan yang keluar masuk [localhost](http://localhost) (127.0.0.1), lalu menjalankan file go tersebut dengan `go run .`, terlihat bahwa permintaan jaringan baru bermunculan dari port default NATS, yaitu `4222`.

# Konfigurasi yang tepat untuk embedding mode

- Agar komunikasi berjalan dalam embedded mode sesuai yang diinginkan, diperlukan dua opsi berikut.
  - Client: opsi `InProcessServer` harus disertakan.
  - Server: flag `DontListen` pada `Server.Options` harus dinyatakan secara eksplisit sebagai `true`.
- Bagian-bagian tersebut tidak didokumentasikan secara resmi, dan asal-usul fitur ini dapat ditelusuri melalui [PR](https://github.com/nats-io/nats-server/pull/2360) berikut.

  > 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

# Contoh kerja untuk 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

```

- Sekarang terlihat bahwa tidak ada network hop tambahan yang terjadi, sesuai dengan yang dimaksud.

# Di balik layar

## TL;DR

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

- Ini adalah sequence diagram yang menunjukkan fungsi-fungsi internal apa saja yang bekerja dan bagaimana cara kerjanya ketika kode tersebut dijalankan dari `main.go`. Inti penjelasannya adalah sebagai berikut.
  - Melalui `DontListen: true`, server melewati client listening phase yang disebut `AcceptLoop`.
  - Jika opsi `InProcessServer` pada Connect client diaktifkan, sebuah in-memory connection dibuat, sebuah pipe dibuat melalui `net.Pipe`, lalu ujung pipe dikembalikan kepada client sebagai tipe `net.Conn`.
  - Client dan server melakukan komunikasi in-process melalui connection tersebut.

## Server

### AcceptLoop

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

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

- Pertama, jika `DontListen` bernilai true, client listening phase yang disebut `AcceptLoop` dilewati.

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

- Sebagai catatan, fungsi AcceptLoop menjalankan proses-proses berikut. Bagian yang berkaitan dengan komunikasi jaringan, seperti `TLS` dan `hostPort`, tidak diperlukan untuk in-process communication sehingga dapat diabaikan.

## 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)
}
```

- Fungsi `Connect` yang menjalankan koneksi antara nats server dan nats client dapat menerima client URL dan connect Option. Struct Options, yang menampung Option-Option tersebut, memiliki field bernama `InProcessServer` dengan tipe interface `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))
```

- Ketika nats client melakukan Connect dan meneruskan `nats.InProcessServer(ns)` ke field `InProcessServer`,

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

- option InProcessServer digantikan dengan 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
	}

	//...
}
```

- Interface tersebut digunakan dalam fungsi `createConn` yang membuat connection. Jika option `InProcessServer` tidak nil (valid), fungsi tersebut menjalankan `InProcessConn` dari InProcessServer pada option tersebut,

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

- Fungsi `InProcessConn` yang diimplementasikan pada server dipanggil.
- Fungsi ini dipanggil oleh `nats.go`, library Go untuk client NATS, ketika `InProcessServer` pada nc (nats connection) tidak nil. Fungsi ini membuat connection (`net.Conn`) dan mengikatnya ke connection pada server.

# Consumer driven interface di 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)

- Desain interface seperti ini sangat merepresentasikan consumer defined interface dan structural typing (duck typing) yang sering disebut dalam Go, sehingga topik ini juga ingin dibahas di sini.

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

- Mari kembali ke kode. Pada client nats.go, field option `InProcessServer` didefinisikan dengan tipe interface `InProcessConnProvider` yang hanya mencakup `InProcessConn`.

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

- Namun, tipe yang dimasukkan ke dalamnya adalah `Server` dari nats-server, yang menjalankan berbagai fungsi selain InProcessConn.
- Hal ini karena dalam konteks ini, yang menjadi perhatian client hanyalah apakah objek tersebut menyediakan interface `InProcessConn` atau tidak; hal lainnya tidak terlalu penting.
- Oleh karena itu, nats.go client hanya mendefinisikan dan menggunakan `InProcessConnProvider`, sebuah consumer defined interface yang hanya memuat fungsi `InProcessConn() (net.Conn, error)`.

# Kesimpulan

- Kami telah membahas secara ringkas embedded mode NATS beserta cara kerjanya, serta consumer defined interface di Go yang dapat dipelajari melalui kode NATS.
- Semoga informasi ini bermanfaat bagi mereka yang menggunakan NATS untuk tujuan serupa, dan dengan itu tulisan ini kami akhiri.