Golang and Event-Driven Architecture
Tags: go, eventdriven, microservices, conference
Published: 2022-04-06 18:15:58

---
title: Golang and Event-Driven Architecture
published: true
description: Overview on how to use Golang in event-driven use-cases
tags: go, eventdriven, microservices, conference
cover_image: https://dev-to-uploads.s3.amazonaws.com/uploads/articles/jl6079wlupxe7xlzczq1.jpeg
---

![Solace Gopher Solly](https://us.v-cdn.net/6031588/uploads/PAQRBSQ510P9/gopher-2bsolly-2bsky.png)

On March 31 2022, I gave a talk at Conf42:Golang about using Go in an event-driven architecture entitled "Gopher in an Event Driven Playground". You can check out the talk here or read along to know more about event-driven systems and how different messaging protocols are used in Go!

{% embed https://www.youtube.com/watch?v=XBkBpFYAHiw&t=76s %}

## What's all the hype about?

For all of us Go enthusiasts out there, we truly understand the beauty of using Go in applications and microservices because of its lightweight, high performance, and elegant syntax, to name just a few (let the debate start! 🔥).

So imagine using your favourite programming language (yes, that is Go) with your favourite architecture (yes, that event-driven architecture) - where do you start? Carry on and read further to explore. 

### Hold on, what is an event-driven architecture (EDA)?

Glad you asked! There are loads of resources online that talk about EDA. At its core, an event-driven architecture involves asynchronous communication between applications via publishing and subscribing to events over an [event broker](https://solace.com/what-is-an-event-broker) using a messaging protocol. 

Some examples of messaging protocols include open standard and open source protocols such as [MQTT](https://mqtt.org/), [AMQP](https://www.amqp.org/), and [JMS](https://www.oracle.com/technical-resources/articles/java/intro-java-message-service.html).

To delve more into event-driven architecture and explore different use-cases, check out this page on [What is Event-Driven Architecture](https://solace.com/what-is-event-driven-architecture).

I also wrote a blog post talking about how I built an event-driven NodeJS app on real-time COVID-19 data streams, highlighting the advantages of event-driven architecture, if you want to explore it further:

{% embed https://dev.to/solacedevs/how-i-built-an-event-driven-nodejs-app-on-realtime-covid-19-data-streams-3i98 %}


## fmt.Println("Show me the CODE!")

Alright, we did a high level overview on what EDA is but let's get to business with some hands-on code. Let me show you how to use:
1. The Paho MQTT library for Golang
1. The PubSub+ Messaging API for Go

Note that all the source code can be found in this open source Github repo:

{% embed https://github.com/TamimiGitHub/conf42-golang-eda %}

### Hands-on 👩‍💻🧑‍💻

Let's start with creating a new directory and initializing go mod. In a new terminal window, execute the following commands:

```go
mkdir funWithEDA && cd "$_"
go mod init GoEDA
```

### MQTT

As per definition: "MQTT is an OASIS standard messaging protocol for the Internet of Things (IoT). It is designed as an extremely lightweight publish/subscribe messaging transport that is ideal for connecting remote devices with a small code footprint and minimal network bandwidth"

The [Eclipse Paho](https://www.eclipse.org/paho/) project provides open source, mainly client side, implementations of MQTT in a variety of programming languages. For today's fun coding session, we will be using the [Eclipse Paho MQTT Go client](https://pkg.go.dev/github.com/eclipse/paho.mqtt.golang#section-readme).

Install the Paho MQTT Go library
```
go get github.com/eclipse/paho.mqtt.golang
```

Create a new file and open it with your favourite IDE. I named my file `go_mqtt.go`.

Initialize the file and import the necessary libraries

```go
package main

import (
	"fmt"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"
)
```

In your `main()` function, start with 
1. Creating a new MQTT client and configure it
1. Add callback functions for
 1. Received Messages 
 1. Successful broker connection
 1. Lost broker connection
1. Subscribe to a topic
1. Publish on a topic

```go
func main() {
	var broker = "public.messaging.solace.cloud"
	var port = 1883
	opts := mqtt.NewClientOptions()
	opts.AddBroker(fmt.Sprintf("tcp://%s:%d", broker, port))
	opts.SetUsername("conf42")
	opts.SetPassword("public")
	opts.SetDefaultPublishHandler(messageHandler)
	opts.OnConnect = connectHandler
	opts.OnConnectionLost = connectLostHandler
	client := mqtt.NewClient(opts)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		panic(token.Error())
	}

	sub(client)
	publish(client)

	client.Disconnect(250)
}
```

We will need to define the callback functions as follows

```go
var messageHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
	fmt.Printf("Received message: %s from topic: %s\n", msg.Payload(), msg.Topic())
}

var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {
	options := client.OptionsReader()
	fmt.Println("Connected to: ", options.Servers())
}

var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
	fmt.Printf("Connect lost: %v", err)
}
``` 
It is important to note that these functions will be triggered "on" particular actions. So for example, the `messageHandler` function will be triggered whenever the MQTT client receives a message via `mqtt.MessageHandler`.

And finally, define your publish and subscribe functions as follows:

```go
func publish(client mqtt.Client) {
	num := 10
	for i := 0; i < num; i++ {
		text := fmt.Sprintf("Message %d", i)
		token := client.Publish("conf42/go", 0, false, text)
		token.Wait()
		time.Sleep(time.Second)
	}
}

func sub(client mqtt.Client) {
	topic := "conf42/#"
	token := client.Subscribe(topic, 1, nil)
	token.Wait()
	fmt.Printf("Subscribed to topic: %s\n", topic)
}
```

And that's it! Run the application and observe the results
```sh
go run go_mqtt.go
```

### Solace PubSub+ Messaging API for Go

Now that you are an expert on messaging concepts with Go, let's take it up a notch and delve into a more advanced messaging API! We'll be using the [Solace PubSub+ Messaging API for Go](https://pkg.go.dev/solace.dev/go/messaging).

Install the Solace Native Go API
```sh
go get solace.dev/go/messaging
```

Create a new file and open it with your favourite IDE. I named my file `solace_publisher.go`.

Import the necessary packages

```go
package main

import (
	"fmt"
	"os"
	"os/signal"
	"strconv"
	"time"

	"solace.dev/go/messaging"
	"solace.dev/go/messaging/pkg/solace/config"
	"solace.dev/go/messaging/pkg/solace/resource"
)
```

Define the configuration parameters to connect to the Solace PubSub+ Broker

```go
// Configuration parameters
	brokerConfig := config.ServicePropertyMap{
		config.TransportLayerPropertyHost:                "tcp://public.messaging.solace.cloud",
		config.ServicePropertyVPNName:                    "public",
		config.AuthenticationPropertySchemeBasicUserName: "conf42",
		config.AuthenticationPropertySchemeBasicPassword: "public",
	}
```

Initialize a messaging service and connect to it

```go
messagingService, err := messaging.NewMessagingServiceBuilder().FromConfigurationProvider(brokerConfig).Build()

if err != nil {
	panic(err)
}

// Connect to the messaging serice
if err := messagingService.Connect(); err != nil {
	panic(err)
}
```
Build a Direct Message Publisher and start it

```go
//  Build a Direct Message Publisher
directPublisher, builderErr := messagingService.CreateDirectMessagePublisherBuilder().Build()
if builderErr != nil {
	panic(builderErr)
}

// Start the publisher
startErr := directPublisher.Start()
if startErr != nil {
	panic(startErr)
}
```

Publish messages in a loop

```go
msgSeqNum := 0

//  Prepare outbound message payload and body
messageBody := "Hello from Conf42"
messageBuilder := messagingService.MessageBuilder().
	WithProperty("application", "samples").
	WithProperty("language", "go")

// Run forever until an interrupt signal is received
go func() {
	for directPublisher.IsReady() {
		msgSeqNum++
		message, err := messageBuilder.BuildWithStringPayload(messageBody + " --> " + strconv.Itoa(msgSeqNum))
		if err != nil {
			panic(err)
		}
		topic := resource.TopicOf("conf42/solace/go/" + strconv.Itoa(msgSeqNum))

		// Publish on dynamic topic with dynamic body
		publishErr := directPublisher.Publish(message, topic)
		if publishErr != nil {
			panic(publishErr)
		}

		fmt.Println("Published message on topic: ", topic.GetName())
		time.Sleep(1 * time.Second)
	}
}()
```

This is the final application.
```go
package main

import (
	"fmt"
	"os"
	"os/signal"
	"strconv"
	"time"

	"solace.dev/go/messaging"
	"solace.dev/go/messaging/pkg/solace/config"
	"solace.dev/go/messaging/pkg/solace/resource"
)

func main() {
	// Configuration parameters
	brokerConfig := config.ServicePropertyMap{
		config.TransportLayerPropertyHost:                "tcp://public.messaging.solace.cloud",
		config.ServicePropertyVPNName:                    "public",
		config.AuthenticationPropertySchemeBasicUserName: "conf42",
		config.AuthenticationPropertySchemeBasicPassword: "public",
	}

	messagingService, err := messaging.NewMessagingServiceBuilder().FromConfigurationProvider(brokerConfig).Build()

	if err != nil {
		panic(err)
	}

	// Connect to the messaging serice
	if err := messagingService.Connect(); err != nil {
		panic(err)
	}

	//  Build a Direct Message Publisher
	directPublisher, builderErr := messagingService.CreateDirectMessagePublisherBuilder().Build()
	if builderErr != nil {
		panic(builderErr)
	}

	// Start the publisher
	startErr := directPublisher.Start()
	if startErr != nil {
		panic(startErr)
	}

	msgSeqNum := 0

	//  Prepare outbound message payload and body
	messageBody := "Hello from Conf42"
	messageBuilder := messagingService.MessageBuilder().
		WithProperty("application", "samples").
		WithProperty("language", "go")

	// Run forever until an interrupt signal is received
	go func() {
		for directPublisher.IsReady() {
			msgSeqNum++
			message, err := messageBuilder.BuildWithStringPayload(messageBody + " --> " + strconv.Itoa(msgSeqNum))
			if err != nil {
				panic(err)
			}
			topic := resource.TopicOf("conf42/solace/go/" + strconv.Itoa(msgSeqNum))

			// Publish on dynamic topic with dynamic body
			publishErr := directPublisher.Publish(message, topic)
			if publishErr != nil {
				panic(publishErr)
			}

			fmt.Println("Published message on topic: ", topic.GetName())
			time.Sleep(1 * time.Second)
		}
	}()

	// Handle OS interrupts
	c := make(chan os.Signal, 1)
	signal.Notify(c, os.Interrupt)

	// Block until an OS interrupt signal is received.
	<-c

	// Terminate the Direct Publisher
	directPublisher.Terminate(1 * time.Second)
	fmt.Println("\nDirect Publisher Terminated? ", directPublisher.IsTerminated())
	// Disconnect the Message Service
	messagingService.Disconnect()
	fmt.Println("Messaging Service Disconnected? ", !messagingService.IsConnected())

}
```


And that's it! Run the publisher as follows
```sh
go run solace_publisher.go
```

Note: You can use the [Solace PubSub+ TryMe](https://docs.solace.com/Cloud/ggs_tryme.htm) tab to connect to the broker and subscribe to any topic you want. Subscribe to topic `conf42/solace/>`


**Bonus!** You can run a subscriber application in another terminal, subscribe to `conf42/solace/>`, and observe the results. You can find more about this on the SolaceSample github org.

**Challenge!** Run a publisher and a subscriber in the same application! 🤯


{% embed https://github.com/SolaceSamples/solace-samples-go %}

### Note on protocol interoperability

Using the Solace PubSub+ Event Broker, you can leverage the protocol translation and interoperability features. Even though the Go API was used to connect to the broker, you can use other Solace APIs and/or open standard protocols to connect to the broker and still have the microservices send and receive messages to each other.


![Solace Protocol translation diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/5pwoa61kble7kq7ss2d3.png)

### Why use Solace native APIs??

You might be wondering, "Why should I use Solace native APIs as opposed to an open-standard open-source messaging protocol like MQTT?" Glad you asked! Check out this blog that talks about the advanced features you can get access to through using Solace's API?

{% embed https://solace.com/blog/solace-messaging-apis-event-mesh-features-performance/ %}

## This is all cool. How can I know more 👀

In summary, here are some quick links

- A [Solace Community post](https://solace.community/discussion/1282/go-lets-go-solace-pubsub-messaging-api-for-go-v1-0-0#latest) listing a bunch of resources
- [Solace PubSub+ Messaging API for Go Samples](https://github.com/SolaceSamples/solace-samples-go)
- [Blog post on the motivation behind a native Solace Go API](https://solace.com/blog/go-messaging-api/)
- [Conf 42 talk on YouTube](https://www.youtube.com/watch?v=XBkBpFYAHiw&t=76s)

If you want to see my colleagues and I during a live streaming session talking about EDA, the Solace PubSub+ Messaging API for Go, and coding (on LIVE television!), check out this event 👀

{% embed https://youtu.be/va2ATJSaUzI %}


## What about you?

I am curious to hear from the go community!
- What is it about Golang that you enjoy?
- Have you ever used Golang in an event-driven application?
- What messaging protocols and/or open standards have you used?
- What message brokers do you use?

Open to all sort of discussions, comments, and questions! 

![PeaceOut Gif](https://media.giphy.com/media/11AuX2SHScQumk/giphy.gif)

And P.S. You can check out the other talks from Conf42 at [conf42.com/golang2022](https://www.conf42.com/golang2022)