diff --git a/Backend.go b/Backend.go new file mode 100644 index 0000000..25f5697 --- /dev/null +++ b/Backend.go @@ -0,0 +1,45 @@ +package main + +import ( + "context" + "fmt" + + "github.com/redis/go-redis/v9" +) + +/* +This file demonstrates a simple **sample backend client** implementation +that listens for messages published on a Redis channel ("Backend"). +The purpose of this file is to serve as a reference for how a backend +service can subscribe to Redis events and process them further (e.g., +forward to clients, trigger business logic, or store in a database). + +Note: This is only an example to illustrate backend-side event handling. +In a real system, this logic can be extended with Propeller gRPC calls, +message routing, persistence, or any other domain-specific functionality. +*/ + +var ctx = context.Background() + +func main() { + // Connect to Redis + rdb := redis.NewClient(&redis.Options{ + Addr: "localhost:6379", // change if Redis is on another host/port + DB: 0, + }) + + // Subscribe to the "backend" channel + subscriber := rdb.Subscribe(ctx, "Backend") + + // Get the channel to receive messages + ch := subscriber.Channel() + + fmt.Println("Backend client subscribed to Redis channel 'backend'...") + + // Listen for messages + for msg := range ch { + fmt.Printf("Received message from Redis: %s\n", msg.Payload) + + // TODO: process event here (e.g. send to clients, store in DB, etc.) + } +} diff --git a/internal/component/apiserver/grpc_handler.go b/internal/component/apiserver/grpc_handler.go index 545052b..513bcfb 100644 --- a/internal/component/apiserver/grpc_handler.go +++ b/internal/component/apiserver/grpc_handler.go @@ -355,7 +355,68 @@ func (ps *PushServer) HandleReceivedPayload(ctx context.Context, srv pushv1.Push ErrorType: "", }, }}}) + case *pushv1.ChannelRequest_ChannelEvent: + + // Referenced and adapted from the SendEventToTopic function for consistent + // context handling, logging, and request publishing flow. + event := receivedRequest.GetChannelEvent().GetEvent() + topic := receivedRequest.GetChannelEvent().GetTopic() + + fmt.Printf("event: %v\n", event) + fmt.Printf("topic: %v\n", topic) + + req := &pushv1.SendEventToTopicRequest{ + Topic: topic, + Event: event, + } + + derivedCtx := context.WithValue(ctx, logger.CtxKeyType("meta"), map[string]string{ + "topic": req.GetTopic(), + "eventName": req.GetEvent().GetName(), + }) + + loggerCtx := context.WithValue(derivedCtx, logger.CtxKey, logger.WithContext(derivedCtx, []logger.CtxKeyType{"meta"})) + + reqModel := push.SendEventToTopicRequest{} + + err := reqModel.PopulateFromProto(loggerCtx, req) + if err != nil { + fmt.Println("Error in Populating from Proto : ",err) + _ = srv.Send(&pushv1.ChannelResponse{Response: &pushv1.ChannelResponse_ChannelEventAck{ChannelEventAck: &pushv1.ChannelEventAck{ + Status: &pushv1.ResponseStatus{ + Success: false, + ErrorCode: "", + Message: map[string]string{"message": err.Error()}, + ErrorType: "", + }, + }}}) + return + } + + err = ps.svc.PublishToTopic(loggerCtx, reqModel) + if err != nil { + fmt.Println("Error in PublisingToTopic : ",err) + _ = srv.Send(&pushv1.ChannelResponse{Response: &pushv1.ChannelResponse_ChannelEventAck{ChannelEventAck: &pushv1.ChannelEventAck{ + Status: &pushv1.ResponseStatus{ + Success: true, + ErrorCode: "", + Message: map[string]string{"message": err.Error()}, + ErrorType: "", + }, + }}}) + return + } + + _ = srv.Send(&pushv1.ChannelResponse{Response: &pushv1.ChannelResponse_ChannelEventAck{ChannelEventAck: &pushv1.ChannelEventAck{ + Status: &pushv1.ResponseStatus{ + Success: true, + ErrorCode: "", + Message: nil, + ErrorType: "", + }, + }}}) } + } func receiveLoop(ctx context.Context, rc chan *pushv1.ChannelRequest, srv pushv1.PushService_ChannelServer) {