06 Eevnt Streaming

Now we will introduces event sourcing using NATS, a lightweight and high-performance messaging system. We'll leverage NATS to create an onboarding task when new users are created and to notify users when they are assigned to a task.

The working branch includes now 3 microservices, let's check

git checkout 05-task-notification-services

Goals

Implement event-driven communication between the User Service and Task Service using NATS.
Create an onboarding task automatically when a new user is registered (User Service publishes event, Task Service consumes it).
Send a notification to a user when they are assigned a task (Task Service publishes event, Notification Service consumes it).

diagram

Add NATS Handler

In the main pkg, create nats.go

mkdir -p pkg/streaming
touch pkg/streaming/nats.go

Add the functionality for the handler

package streaming

import (
    "encoding/json"

    "github.com/nats-io/nats.go"
    "go.uber.org/zap"
)

type EventHandler interface {
    HandleEvent(event map[string]interface{}) error
}

type NATSSubscriber struct {
    nc           *nats.Conn
    eventHandler EventHandler
    logger       *zap.Logger
}

func NewNATSSubscriber(nc *nats.Conn, eventHandler EventHandler, logger *zap.Logger) *NATSSubscriber {
    return &NATSSubscriber{
        nc:           nc,
        eventHandler: eventHandler,
        logger:       logger,
    }
}

func (s *NATSSubscriber) Subscribe(subject string) error {
    _, err := s.nc.Subscribe(subject, func(m *nats.Msg) {
        var event map[string]interface{}
        if err := json.Unmarshal(m.Data, &event); err != nil {
            s.logger.Error("Failed to unmarshal event", zap.Error(err))
            return
        }

        s.logger.Info("Received event", zap.Any("event", event))

        if err := s.eventHandler.HandleEvent(event); err != nil {
            s.logger.Error("Failed to handle event", zap.Error(err))
        }
    })
    return err
}

go mod tidy

User Service

Initialize NATS connection: Update cmd/command/main.go (line:50)

// ... other imports
import "github.com/nats-io/nats.go"

// ... 
func main() {
    // ... (Other initialization)

    // Publish event to NATS
    natsService := viper.GetString("NATS_SERVICE")
    natsPort := viper.GetString("NATS_PORT")
    logger.Info("Connecting to NATS")
    nc, err := nats.Connect(fmt.Sprintf("nats://%s:%s", natsService, natsPort))
    if err != nil {
        log.Fatal("Failed to connect to NATS", err)
    }
    defer nc.Close()

    // ... (Rest of your main function)
    userHandler := handlers.NewUserCommandHandler(userCommand, nc, logger)

}

Publish a user_created event when a new user is created:

// user-service/internal/handlers/command_user.go
package handlers

// ...
import (
    "encoding/json"
    // ...
)

// ... update UserCommandHandler to accept NATS Client
type UserCommandHandler struct {
    UserCommand command.UserCommand
    logger      *zap.Logger
    nc          *nats.Conn // <----- add this
}

func NewUserCommandHandler(userCommand command.UserCommand, nc *nats.Conn, logger *zap.Logger) *UserCommandHandler {
    return &UserCommandHandler{
        UserCommand: userCommand,
        logger:      logger,
        nc:          nc, // <----- add this
    }
}

// ...
func (h *UserCommandHandler) createUser(c *fiber.Ctx) error {
    // ... (Logic to create a new user) ...


    eventData := map[string]interface{}{
        "user_id": id,
        "action":  "user_created",
    }
    eventBytes, _ := json.Marshal(eventData)
    err = h.nc.Publish("user.events", eventBytes)
    if err != nil {
        h.logger.Error("Failed to publish user created event", zap.Error(err))
        return c.Status(fiber.StatusInternalServerError).SendString(err.Error())
    }

    return c.Status(fiber.StatusCreated).JSON(user)
}

Update create user command to return created id

//user-service/internal/repository/command/user.go
func (c *MySQLUserCommand) CreateUser(ctx context.Context, user *models.User) (int64, error) {
    res, err := c.Conn.ExecContext(ctx, "INSERT INTO users (name, email, password) VALUES (?, ?, ?)", user.Name, user.Email, user.Password)
    if err != nil {
        return -1, err
    }
    return res.LastInsertId()
}

Task Service

Initialize NATS connection:

// ... other imports
import "github.com/nats-io/nats.go"

// ...
func main() {
    // ... (other initialization)

    // Connect to NATS
    natsService := viper.GetString("NATS_SERVICE")
    natsPort := viper.GetString("NATS_PORT")
    logger.Info("Connecting to NATS")
    nc, err := nats.Connect(fmt.Sprintf("nats://%s:%s", natsService, natsPort))
    if err != nil {
        log.Fatal("Failed to connect to NATS", err)
    }
    defer nc.Close()

    eventHandler := internalstream.NewUserEventHandler(taskCommand, logger)

    // Initialize the NATS subscriber
    subscriber := streaming.NewNATSSubscriber(nc, eventHandler, logger)
    if err := subscriber.Subscribe("user.events"); err != nil {
        logger.Fatal("Failed to subscribe to NATS", zap.Error(err))
    }

    // Initialize the handler
    logger.Info("Initializing task command handler")
    taskHandler := handlers.NewTaskCommandHandler(taskCommand, nc, logger)

    // ... (Rest of your main function)
}

Subscribe to the user_created event to create an onboarding task:

mkdir -p task-service/internal/streaming
touch task-service/internal/streaming/user.go

// task-service/internal/streaming/user.go
package streaming

import (
    "context"

    "github.com/rfashwall/task-service/internal/models"
    "github.com/rfashwall/task-service/internal/repository/command"
    "go.uber.org/zap"
)

type UserEventHandler struct {
    taskCommand command.TaskCommand
    logger      *zap.Logger
}

func NewUserEventHandler(taskCommand command.TaskCommand, logger *zap.Logger) *UserEventHandler {
    return &UserEventHandler{
        taskCommand: taskCommand,
        logger:      logger,
    }
}

func (h *UserEventHandler) HandleEvent(event map[string]interface{}) error {
    switch event["action"] {
    case "user_created":
        h.logger.Info("User created event received")
        userID := int(event["user_id"].(float64))
        h.logger.Info("Creating onboarding task", zap.Int("user_id", userID), zap.Any("event", event))
        task := &models.Task{
            UserID:      userID,
            Title:       "Onboarding Task",
            Description: "Complete your onboarding process",
            Status:      "Pending",
            AssigneeID:  userID,
        }
        h.logger.Info("Creating onboarding task", zap.Int("user_id", userID), zap.Any("task", task))
        return h.taskCommand.CreateTask(context.Background(), task)
    default:
        h.logger.Warn("Unknown event action", zap.Any("event", event))
    }

    return nil
}

Publish a task_assigned event when a task is assigned to a user:

// task-service/internal/handlers/command-task.go
// ... 
func (h *TaskCommandHandler) assignTask(c *fiber.Ctx) error {
    // ... (Logic to assign task) ...

eventData := map[string]interface{}{
        "user_id": req.AssigneeID,
        "action":  "task_assigned",
    }
    eventBytes, _ := json.Marshal(eventData)
    err = h.nc.Publish("task.events", eventBytes)
    if err != nil {
        h.logger.Error("Failed to publish task assigned event", zap.Error(err))
        return c.Status(fiber.StatusInternalServerError).SendString(err.Error())
    }

    return c.SendStatus(fiber.StatusOK)
}

// ...

Inter-service communication

We want the task-command-service to check if user exists before creating a task.

To achieve this, we will add a user service handler in task service to fetch the user data using Synchronous Communication. This involves direct communication between services, where one service makes a request to another and waits for a response.

Add User Handler

mkdir -p task-service/internal/service
touch task-service/internal/service/user.go

package service

import (
    "context"
    "encoding/json"
    "fmt"
    "net/http"

    "github.com/rfashwall/task-service/internal/models"
)

type UserService interface {
    GetUserByID(ctx context.Context, id int) (*models.User, error)
}

type HTTPUserService struct {
    BaseURL string
}

func NewHTTPUserService(baseURL string) *HTTPUserService {
    return &HTTPUserService{BaseURL: baseURL}
}

func (s *HTTPUserService) GetUserByID(ctx context.Context, id int) (*models.User, error) {
    url := fmt.Sprintf("%s/users/%d", s.BaseURL, id)
    req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
    if err != nil {
        return nil, err
    }

    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        return nil, fmt.Errorf("failed to get user: status %d", resp.StatusCode)
    }

    var user models.User
    if err := json.NewDecoder(resp.Body).Decode(&user); err != nil {
        return nil, err
    }

    return &user, nil
}

Update Task Command Handler to accept UserService Handler

    type TaskCommandHandler struct {
    TaskCommand command.TaskCommand
    nc          *nats.Conn
    userService service.UserService // <-- add
    logger      *zap.Logger
}

func NewTaskCommandHandler(taskCommand command.TaskCommand, nc *nats.Conn, userService service.UserService, logger *zap.Logger) *TaskCommandHandler {
    return &TaskCommandHandler{
        TaskCommand: taskCommand,
        nc:          nc,
        userService: userService, // <-- add
        logger:      logger,
    }
}

Update cmd/command/main.go

// ...

// Add user service config
userSvcHost := viper.GetString("USER_SERVICE_HOST")
userSvcPort := viper.GetString("USER_SERVICE_PORT")

// ...


	// Initialize the repository
taskCommand := command.NewMySQLTaskCommand(conn)

userService := service.NewHTTPUserService(fmt.Sprintf("%s:%s", userSvcHost, userSvcPort)) // <---- add this line  to init a user handler 



logger.Info("Initializing task command handler")
taskHandler := handlers.NewTaskCommandHandler(taskCommand, nc, userService, logger) // <--- pass user service handler

// ...

Notification Service

This service will be responsible for sending notifications to users

Init files

mkdir -p notification-service/internal/{streaming,utils}
touch notification-service/internal/streaming/task.go
touch notification-service/internal/utils/notifier.go

Update notification-service/cmd/main.go:

package main

import (
    "fmt"
    "log"

    "github.com/gofiber/fiber/v2"
    "github.com/gofiber/fiber/v2/middleware/healthcheck"
    "github.com/gofiber/fiber/v2/middleware/logger"
    "github.com/nats-io/nats.go"
    "github.com/rfashwall/go-task/pkg/middleware"
    "github.com/rfashwall/go-task/pkg/streaming"
    "github.com/rfashwall/go-task/pkg/utils"
    internalstream "github.com/rfashwall/notification-service/internal/streaming"
    nutils "github.com/rfashwall/notification-service/internal/utils"
    "github.com/spf13/viper"
    "go.uber.org/zap"
)

func main() {
    shutdown := utils.InitTracer()
    defer shutdown()

    envPrefix := "NOTIFICATION_SERVICE"

    viper.SetEnvPrefix(envPrefix)
    viper.AutomaticEnv()

    serviceName := viper.GetString("SERVICE_NAME")
    servicePort := viper.GetString("SERVICE_PORT")

    app := fiber.New()
    app.Use(logger.New())
    app.Use(healthcheck.New())
    app.Use(middleware.TracingMiddleware(serviceName))

    logger, err := zap.NewDevelopment()
    if err != nil {
        log.Fatal(err)
    }
    defer logger.Sync()

    // Publish event to NATS
    natsService := viper.GetString("NATS_SERVICE")
    natsPort := viper.GetString("NATS_PORT")
    logger.Info("Connecting to NATS")
    nc, err := nats.Connect(fmt.Sprintf("nats://%s:%s", natsService, natsPort))
    if err != nil {
        log.Fatal("Failed to connect to NATS", err)
    }
    defer nc.Close()

    // Send email notification
    emailNotifier := nutils.NewEmailNotifier()
    eventHandler := internalstream.NewTaskEventHandler(emailNotifier, logger)

    // Initialize the NATS subscriber
    subscriber := streaming.NewNATSSubscriber(nc, eventHandler, logger)
    if err := subscriber.Subscribe("task.events"); err != nil {
        logger.Fatal("Failed to subscribe to NATS", zap.Error(err))
    }

    logger.Info(fmt.Sprintf("Listening on port %s", servicePort))
    log.Fatal(app.Listen(fmt.Sprintf(":%s", servicePort)))
}

Add Task Events Handler

package streaming

import (
    "github.com/rfashwall/notification-service/internal/utils"
    "go.uber.org/zap"
)

type TaskEventHandler struct {
    notifier utils.Notifier
    logger   *zap.Logger
}

func NewTaskEventHandler(notifier utils.Notifier, logger *zap.Logger) *TaskEventHandler {
    return &TaskEventHandler{
        notifier: notifier,
        logger:   logger,
    }
}

func (h *TaskEventHandler) HandleEvent(event map[string]interface{}) error {
    switch event["action"] {
    case "task_created":
        h.logger.Info("Task created event received")
        return h.notifier.Send("", "userID")
    case "task_assigned":
        h.logger.Info("Task assigned event received")
        return h.notifier.Send("", "userID")
    default:
        h.logger.Warn("Unknown event action", zap.Any("event", event))
    }

    return nil
}

Add Notifier Dummy Implementation

package utils

import (
    "fmt"
)

type NotificationType string

const (
    Email NotificationType = "Email"
)

type Notifier interface {
    Send(recipient, message string) error
}

type EmailNotifier struct{}

func NewEmailNotifier() Notifier {
    return &EmailNotifier{}
}
func (e *EmailNotifier) Send(recipient, message string) error {
    fmt.Printf("Sending email to %s: %s\n", recipient, message)
    return nil // Simulate sending email
}

If imports didn't work execute in each service bash go get github.com/rfashwall/go-task@6b9164a19729fa476118c5c2cb4041c601a87f0a go mod tidy

Update Docker Compose

Replace with the below code

version: '3.8'

services:
mysql-server:
    image: mysql:latest
    container_name: mysql-server
    environment:
        MYSQL_ROOT_PASSWORD: root
        MYSQL_DATABASE: user_tracker
    ports:
        - "3306:3306"
    volumes:
        - mysql-data:/var/lib/mysql
        - ./init.sh:/docker-entrypoint-initdb.d/init.sh
    healthcheck:
        test: ["CMD-SHELL", "mysqladmin ping -h localhost"]
        interval: 10s
        retries: 5
        start_period: 30s
        timeout: 10s

user-command-service:
    build:
    context: ./user-service
    dockerfile: Dockerfile.command
    container_name: user-command-service
    environment:
    - USER_SERVICE_DB_HOST=mysql-server 
    - USER_SERVICE_DB_PORT=3306
    - USER_SERVICE_DB_USER=root
    - USER_SERVICE_DB_PASS=root
    - USER_SERVICE_DB_NAME=user_tracker
    - USER_SERVICE_SERVICE_NAME=user-command-service
    - USER_SERVICE_SERVICE_PORT=3001
    - USER_SERVICE_NATS_PORT=4222
    - USER_SERVICE_NATS_SERVICE=nats
    ports:
    - "3001:3001"
    depends_on:
    mysql-server:
        condition: service_healthy

user-query-service:
    build:
    context: ./user-service
    dockerfile: Dockerfile.query
    container_name: user-query-service
    ports:
    - "3000:3000"
    environment:
    - USER_SERVICE_DB_HOST=mysql-server 
    - USER_SERVICE_DB_PORT=3306
    - USER_SERVICE_DB_USER=root
    - USER_SERVICE_DB_PASS=root
    - USER_SERVICE_DB_NAME=user_tracker
    - USER_SERVICE_SERVICE_NAME=user-query-service
    - USER_SERVICE_SERVICE_PORT=3000
    depends_on:
    mysql-server:
        condition: service_healthy

task-command-service:
    build:
    context: ./task-service
    dockerfile: Dockerfile.command
    container_name: task-command-service
    ports:
    - "3002:3002"
    environment:
    - TASK_SERVICE_DB_HOST=mysql-server 
    - TASK_SERVICE_DB_PORT=3306
    - TASK_SERVICE_DB_USER=root
    - TASK_SERVICE_DB_PASS=root
    - TASK_SERVICE_DB_NAME=task_tracker
    - TASK_SERVICE_SERVICE_NAME=task-command-service
    - TASK_SERVICE_SERVICE_PORT=3002
    - TASK_SERVICE_NATS_PORT=4222
    - TASK_SERVICE_NATS_SERVICE=nats
    - TASK_SERVICE_USER_SERVICE_HOST=user-query-service
    - TASK_SERVICE_USER_SERVICE_PORT=3000
    depends_on:
    mysql-server:
        condition: service_healthy

task-query-service:
    build:
    context: ./task-service
    dockerfile: Dockerfile.query
    container_name: task-query-service
    ports:
    - "3003:3003"
    environment:
    - TASK_SERVICE_DB_HOST=mysql-server 
    - TASK_SERVICE_DB_PORT=3306
    - TASK_SERVICE_DB_USER=root
    - TASK_SERVICE_DB_PASS=root
    - TASK_SERVICE_DB_NAME=task_tracker
    - TASK_SERVICE_SERVICE_NAME=task-query-service
    - TASK_SERVICE_SERVICE_PORT=3003
    depends_on:
    mysql-server:
        condition: service_healthy

notification-service:
    build:
    context: ./notification-service
    dockerfile: Dockerfile
    container_name: notification-service
    ports:
    - "3004:3004"
    environment:
    - NOTIFICATION_SERVICE_SERVICE_NAME=notification-service
    - NOTIFICATION_SERVICE_SERVICE_PORT=3004
    - NOTIFICATION_SERVICE_NATS_PORT=4222
    - NOTIFICATION_SERVICE_NATS_SERVICE=nats
    depends_on:
    - nats

nats:
    image: nats:latest
    container_name: nats
    ports:
    - "4222:4222"
    - "6222:6222"
    - "8222:8222"

volumes:
mysql-data:

Navigation Menu

Repository navigation

06 Eevnt Streaming

Goals

Add NATS Handler

User Service

Task Service

Inter-service communication

Notification Service

Update Docker Compose

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally