» Go: Building Event-Driven Microservices with Kafka » 4. Consumer: Recommendation Service » 4.2 Event Consumer
Event Consumer
Since the recommendation service‘s consumer has so many things in common with the trend service‘s counterpart, let‘s try to reuse them by moving those common parts into the higher level folders.
Restructure files
- Tune service/trend/domain/gateway/trend_manager.go:
@@ -9,16 +9,8 @@ import (
"literank.com/event-books/domain/model"
)
-type ConsumeCallback func(key, value []byte) error
-
// TrendManager manages all trends
type TrendManager interface {
CreateTrend(ctx context.Context, t *model.Trend) (uint, error)
TopTrends(ctx context.Context, pageSize uint) ([]*model.Trend, error)
}
-
-// TrendEventConsumer consumes trend events
-type TrendEventConsumer interface {
- ConsumeEvents(ctx context.Context, callback ConsumeCallback)
- Stop() error
-}
Move the common TrendEventConsumer and ConsumeCallback types out of this file.
- Put those types in domain/gateway/event_consumer.go:
package gateway
import "context"
type ConsumeCallback func(key, value []byte) error
// EventConsumer consumes all kinds of events
type EventConsumer interface {
ConsumeEvents(ctx context.Context, callback ConsumeCallback)
Stop() error
}
- Rename infrastructure/mq/kafka.go to infrastructure/mq/kafka_producer.go.
- Move service/trend/infrastructure/mq/kafka.go to infrastructure/mq/kafka_consumer.go.
func NewKafkaConsumer(brokers []string, topic, groupID string) (*KafkaConsumer, error) {
Add a new parameter named groupID for the consumer group.
- Add config items in service/trend/infrastructure/config/config.go:
@@ -26,4 +26,5 @@ type CacheConfig struct {
type MQConfig struct {
Brokers []string `json:"brokers" yaml:"brokers"`
Topic string `json:"topic" yaml:"topic"`
+ GroupID string `json:"group_id" yaml:"group_id"`
}
- Put values in service/trend/config.yml:
@@ -7,4 +7,5 @@ cache:
mq:
brokers:
- localhost:9094
topic: "lr-book-searches"
+ group_id: "trend-svr"
- Tune import pathes in service/trend/application/consumer/trend.go:
@@ -7,16 +7,17 @@ import (
"context"
"encoding/json"
+ topgw "literank.com/event-books/domain/gateway"
"literank.com/event-books/domain/model"
"literank.com/event-books/service/trend/domain/gateway"
)
type TrendConsumer struct {
trendManager gateway.TrendManager
- eventConsumer gateway.TrendEventConsumer
+ eventConsumer topgw.EventConsumer
}
-func NewTrendConsumer(t gateway.TrendManager, e gateway.TrendEventConsumer) *TrendConsumer {
+func NewTrendConsumer(t gateway.TrendManager, e topgw.EventConsumer) *TrendConsumer {
return &TrendConsumer{trendManager: t, eventConsumer: e}
}
@@ -33,6 +34,6 @@ func (c *TrendConsumer) Start(ctx context.Context) {
})
}
-func (c *TrendConsumer) EventConsumer() gateway.TrendEventConsumer {
+func (c *TrendConsumer) EventConsumer() topgw.EventConsumer {
return c.eventConsumer
}
- Tune import paths in service/trend/application/wire_helper.go:
@@ -4,10 +4,11 @@ Package application provides all common structures and functions of the applicat
package application
import (
+ topgw "literank.com/event-books/domain/gateway"
+ "literank.com/event-books/infrastructure/mq"
"literank.com/event-books/service/trend/domain/gateway"
"literank.com/event-books/service/trend/infrastructure/cache"
"literank.com/event-books/service/trend/infrastructure/config"
- "literank.com/event-books/service/trend/infrastructure/mq"
)
// WireHelper is the helper for dependency injection
@@ -19,7 +20,7 @@ type WireHelper struct {
// NewWireHelper constructs a new WireHelper
func NewWireHelper(c *config.Config) (*WireHelper, error) {
kv := cache.NewRedisCache(c.Cache.Address, c.Cache.Password, c.Cache.DB)
- consumer, err := mq.NewKafkaConsumer(c.MQ.Brokers, c.MQ.Topic)
+ consumer, err := mq.NewKafkaConsumer(c.MQ.Brokers, c.MQ.Topic, c.MQ.GroupID)
if err != nil {
return nil, err
}
@@ -35,6 +36,6 @@ func (w *WireHelper) TrendManager() gateway.TrendManager {
}
// TrendEventConsumer returns an instance of TrendEventConsumer
-func (w *WireHelper) TrendEventConsumer() gateway.TrendEventConsumer {
+func (w *WireHelper) TrendEventConsumer() topgw.EventConsumer {
return w.consumer
}
Tune the producer‘s events in the web service
Add the userID parameter in service/web/application/executor/book_operator.go:
@@ -36,18 +36,19 @@ func (o *BookOperator) CreateBook(ctx context.Context, b *model.Book) (*model.Bo
}
// GetBooks gets a list of books by offset and keyword, and caches its result if needed
-func (o *BookOperator) GetBooks(ctx context.Context, offset int, query string) ([]*model.Book, error) {
+func (o *BookOperator) GetBooks(ctx context.Context, offset int, userID, query string) ([]*model.Book, error) {
books, err := o.bookManager.GetBooks(ctx, offset, query)
if err != nil {
return nil, err
}
- // Send search query and its results
+ // Send a user's search query and its results
if query != "" {
+ k := query + ":" + userID
jsonData, err := json.Marshal(books)
if err != nil {
return nil, fmt.Errorf("failed to send event due to %w", err)
}
- o.mqHelper.SendEvent(query, jsonData)
+ o.mqHelper.SendEvent(k, jsonData)
}
return books, nil
}
Read the cookie value and pass in as the userID in service/web/adapter/router.go:
@@ -6,8 +6,10 @@ package adapter
import (
"fmt"
"log"
+ "math/rand"
"net/http"
"strconv"
+ "time"
"github.com/gin-gonic/gin"
@@ -20,6 +22,7 @@ import (
const (
fieldOffset = "o"
fieldQuery = "q"
+ fieldUID = "uid"
)
// RestHandler handles all restful requests
@@ -54,8 +57,14 @@ func MakeRouter(templates_pattern string, remote *config.RemoteServiceConfig, wi
// Render and show the index page
func (r *RestHandler) indexPage(c *gin.Context) {
+ userID, err := c.Cookie(fieldUID)
+ if err != nil {
+ // Doesn't exist, make a new one
+ userID = randomString(5)
+ c.SetCookie(fieldUID, userID, 3600*24*30, "/", "", false, false)
+ }
q := c.Query(fieldQuery)
- books, err := r.bookOperator.GetBooks(c, 0, q)
+ books, err := r.bookOperator.GetBooks(c, 0, userID, q)
if err != nil {
c.String(http.StatusNotFound, "failed to get books")
return
@@ -87,7 +96,7 @@ func (r *RestHandler) getBooks(c *gin.Context) {
}
offset = value
}
- books, err := r.bookOperator.GetBooks(c, offset, c.Query(fieldQuery))
+ books, err := r.bookOperator.GetBooks(c, offset, "", c.Query(fieldQuery))
if err != nil {
fmt.Printf("Failed to get books: %v\n", err)
c.JSON(http.StatusNotFound, gin.H{"error": "failed to get books"})
@@ -112,3 +121,14 @@ func (r *RestHandler) createBook(c *gin.Context) {
}
c.JSON(http.StatusCreated, book)
}
+
+func randomString(length int) string {
+ source := rand.NewSource(time.Now().UnixNano())
+ random := rand.New(source)
+ const charset = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
+ result := make([]byte, length)
+ for i := range result {
+ result[i] = charset[random.Intn(len(charset))]
+ }
+ return string(result)
+}
Tune the consumer in the trend service
Since the key‘s format has changed, we need to adapt to that in the trend service.
Update service/trend/application/consumer/trend.go:
@@ -6,6 +6,7 @@ package consumer
import (
"context"
"encoding/json"
+ "strings"
topgw "literank.com/event-books/domain/gateway"
"literank.com/event-books/domain/model"
@@ -23,8 +24,10 @@ func NewTrendConsumer(t gateway.TrendManager, e topgw.EventConsumer) *TrendConsu
func (c *TrendConsumer) Start(ctx context.Context) {
c.eventConsumer.ConsumeEvents(ctx, func(key, data []byte) error {
+ parts := strings.Split(string(key), ":")
+ query := parts[0]
t := &model.Trend{
- Query: string(key),
+ Query: query,
}
if err := json.Unmarshal(data, &t.Books); err != nil {
return err
Recommendation Service‘s consumer
Add service/recommendation/domain/model/interest.go:
package model
// Interest represents a user's interest in a book.
type Interest struct {
UserID string `json:"user_id" bson:"user_id"`
Title string `json:"title"`
Author string `json:"author"`
Score float32 `json:"score"`
}
Add service/recommendation/domain/gateway/interest_manager.go:
/*
Package gateway contains all domain gateways.
*/
package gateway
import (
"context"
"literank.com/event-books/service/recommendation/domain/model"
)
// InterestManager manages all interests
type InterestManager interface {
IncreaseInterest(ctx context.Context, i *model.Interest) error
ListInterests(ctx context.Context, userID string) ([]*model.Interest, error)
}
Install mongo dependencies:
go get -u go.mongodb.org/mongo-driver/mongo
Prepare the mongoDB database:
- Install mongoDB on your machine and start it.
- Create a database named
lr_event_rec.- Create necessary indexes for your collections.
Implement the InterestManager in service/recommendation/infrastructure/database/mongo.go:
/*
Package database does all db persistence implementations.
*/
package database
import (
"context"
"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
"literank.com/event-books/service/recommendation/domain/model"
)
const (
collReview = "interests"
)
// MongoPersistence runs all mongoDB operations
type MongoPersistence struct {
db *mongo.Database
coll *mongo.Collection
pageSize int
}
// NewMongoPersistence constructs a new MongoPersistence
func NewMongoPersistence(mongoURI, dbName string, pageSize int) (*MongoPersistence, error) {
client, err := mongo.Connect(context.Background(), options.Client().ApplyURI(mongoURI))
if err != nil {
return nil, err
}
db := client.Database(dbName)
coll := db.Collection(collReview)
return &MongoPersistence{db, coll, pageSize}, nil
}
// GetReview gets a review by ID
func (m *MongoPersistence) IncreaseInterest(ctx context.Context, i *model.Interest) error {
filter := bson.M{
"user_id": i.UserID,
"title": i.Title,
"author": i.Author,
}
update := bson.M{"$inc": bson.M{"score": 1}}
opts := options.Update().SetUpsert(true)
if _, err := m.coll.UpdateOne(ctx, filter, update, opts); err != nil {
return err
}
return nil
}
// ListInterests lists user interests by a use id
func (m *MongoPersistence) ListInterests(ctx context.Context, userID string) ([]*model.Interest, error) {
filter := bson.M{"user_id": userID}
opts := options.Find()
opts.SetSort(bson.M{"score": -1})
opts.SetLimit(int64(m.pageSize))
cursor, err := m.coll.Find(ctx, filter, opts)
if err != nil {
return nil, err
}
defer cursor.Close(ctx)
interests := make([]*model.Interest, 0)
if err := cursor.All(ctx, &interests); err != nil {
return nil, err
}
return interests, nil
}
Add service/recommendation/infrastructure/config/config.go:
/*
Package config provides config structures and parse funcs.
*/
package config
// Config is the global configuration.
type Config struct {
App ApplicationConfig `json:"app" yaml:"app"`
DB DBConfig `json:"db" yaml:"db"`
MQ MQConfig `json:"mq" yaml:"mq"`
}
// DBConfig is the configuration of databases.
type DBConfig struct {
MongoURI string `json:"mongo_uri" yaml:"mongo_uri"`
MongoDBName string `json:"mongo_db_name" yaml:"mongo_db_name"`
}
// ApplicationConfig is the configuration of main app.
type ApplicationConfig struct {
Port int `json:"port" yaml:"port"`
PageSize int `json:"page_size" yaml:"page_size"`
}
// MQConfig is the configuration of message queues.
type MQConfig struct {
Brokers []string `json:"brokers" yaml:"brokers"`
Topic string `json:"topic" yaml:"topic"`
GroupID string `json:"group_id" yaml:"group_id"`
}
Add service/recommendation/config.yml:
app:
port: 8082
page_size: 10
db:
mongo_uri: "mongodb://localhost:27017"
mongo_db_name: "lr_event_rec"
mq:
brokers:
- localhost:9094
topic: "lr-book-searches"
group_id: "rec-svr"
Add service/recommendation/application/consumer/interest.go:
/*
Package consumer handles event-trigger style business logic.
*/
package consumer
import (
"context"
"encoding/json"
"strings"
topgw "literank.com/event-books/domain/gateway"
topmodel "literank.com/event-books/domain/model"
"literank.com/event-books/service/recommendation/domain/gateway"
"literank.com/event-books/service/recommendation/domain/model"
)
type InterestConsumer struct {
interestManager gateway.InterestManager
eventConsumer topgw.EventConsumer
}
func NewInterestConsumer(t gateway.InterestManager, e topgw.EventConsumer) *InterestConsumer {
return &InterestConsumer{interestManager: t, eventConsumer: e}
}
func (c *InterestConsumer) Start(ctx context.Context) {
c.eventConsumer.ConsumeEvents(ctx, func(key, data []byte) error {
parts := strings.Split(string(key), ":")
if len(parts) == 1 {
// No userID, ignore it
return nil
}
var books []*topmodel.Book
if err := json.Unmarshal(data, &books); err != nil {
return err
}
userID := parts[1]
for _, book := range books {
i := &model.Interest{
UserID: userID,
Title: book.Title,
Author: book.Author,
}
if err := c.interestManager.IncreaseInterest(ctx, i); err != nil {
return err
}
}
return nil
})
}
func (c *InterestConsumer) EventConsumer() topgw.EventConsumer {
return c.eventConsumer
}
Add service/recommendation/application/wire_helper.go:
/*
Package application provides all common structures and functions of the application layer.
*/
package application
import (
topgw "literank.com/event-books/domain/gateway"
"literank.com/event-books/infrastructure/mq"
"literank.com/event-books/service/recommendation/domain/gateway"
"literank.com/event-books/service/recommendation/infrastructure/config"
"literank.com/event-books/service/recommendation/infrastructure/database"
)
// WireHelper is the helper for dependency injection
type WireHelper struct {
noSQLPersistence *database.MongoPersistence
consumer *mq.KafkaConsumer
}
// NewWireHelper constructs a new WireHelper
func NewWireHelper(c *config.Config) (*WireHelper, error) {
mdb, err := database.NewMongoPersistence(c.DB.MongoURI, c.DB.MongoDBName, c.App.PageSize)
if err != nil {
return nil, err
}
consumer, err := mq.NewKafkaConsumer(c.MQ.Brokers, c.MQ.Topic, c.MQ.GroupID)
if err != nil {
return nil, err
}
return &WireHelper{
noSQLPersistence: mdb, consumer: consumer,
}, nil
}
// InterestManager returns an instance of InterestManager
func (w *WireHelper) InterestManager() gateway.InterestManager {
return w.noSQLPersistence
}
// TrendEventConsumer returns an instance of TrendEventConsumer
func (w *WireHelper) TrendEventConsumer() topgw.EventConsumer {
return w.consumer
}
Add service/recommendation/main.go:
package main
import (
"context"
"log"
"os"
"os/signal"
"syscall"
"literank.com/event-books/infrastructure/parser"
"literank.com/event-books/service/recommendation/application"
"literank.com/event-books/service/recommendation/application/consumer"
"literank.com/event-books/service/recommendation/infrastructure/config"
)
const configFileName = "config.yml"
func main() {
// Read the config
c, err := parser.Parse[config.Config](configFileName)
if err != nil {
panic(err)
}
// Prepare dependencies
wireHelper, err := application.NewWireHelper(c)
if err != nil {
panic(err)
}
// Run the consumer
tc := consumer.NewInterestConsumer(wireHelper.InterestManager(), wireHelper.TrendEventConsumer())
eventConsumer := tc.EventConsumer()
go func() {
// Shutdown signals
stopAll := make(chan os.Signal, 1)
signal.Notify(stopAll, syscall.SIGINT, syscall.SIGTERM)
<-stopAll
if err := eventConsumer.Stop(); err != nil {
log.Panicf("Failed to close consumer group: %v", err)
}
}()
tc.Start(context.Background())
}
Start the recommendation consumer:
# in service/recommendation
go run main.go
You should see something similar to this:
2024/04/05 23:56:59 Started to consume events...
From another terminal, restart your web service and try searching with terms like “love“ and “war“ a couple of times.
Then, you will have records like below in your mongoDB:
lr_event_rec> db.interests.find();
[
{
_id: ObjectId('66101a027a7027627b271269'),
author: 'Jane Austen',
title: 'Pride and Prejudice',
user_id: 'YVM2P',
score: 1
},
{
_id: ObjectId('66101a027a7027627b27126b'),
author: 'Leo Tolstoy',
title: 'War and Peace',
user_id: 'YVM2P',
score: 2
},
{
_id: ObjectId('66101a437a7027627b271299'),
author: 'Homer',
title: 'The Odyssey',
user_id: 'YVM2P',
score: 1
}
]
That means your consumer is doing well.
Loading...
> code result goes here