moved active evt into event proper

xigxog · Oct 18, 2023 · 0dfa80b · 0dfa80b
1 parent e7ed6b1
commit 0dfa80b
Show file tree

Hide file tree

Showing 14 changed files with 638 additions and 671 deletions.
diff --git a/components/broker/engine/broker.go b/components/broker/engine/broker.go
diff --git a/components/broker/engine/grpc_server.go b/components/broker/engine/grpc_server.go
@@ -137,14 +137,14 @@ func (srv *GRPCServer) Subscribe(stream grpc.Broker_SubscribeServer) error {
 		return log.ErrorN("component registration failed: %v", err)
 	}
 
-	sendEvt := func(mEvt *kubefox.MatchedEvent) error {
+	sendEvt := func(evt *LiveEvent) error {
 		// Protect the stream from being called by multiple threads.
 		sendMutex.Lock()
 		defer sendMutex.Unlock()
 
-		srv.log.WithEvent(mEvt.Event).Debug("send event")
+		srv.log.WithEvent(evt.Event).Debug("send event")
 
-		if err := stream.Send(mEvt); err != nil {
+		if err := stream.Send(evt.MatchedEvent); err != nil {
 			return fmt.Errorf("%w: %v", ErrComponentGone, err)
 		}
 		return nil
@@ -197,9 +197,10 @@ func (srv *GRPCServer) Subscribe(stream grpc.Broker_SubscribeServer) error {
 
 			log = srv.log.WithEvent(evt)
 			log.Debug("receive event")
-			err = srv.brk.RecvEvent(&ReceivedEvent{
-				ActiveEvent: kubefox.StartEvent(evt),
-				Receiver:    ReceiverGRPCServer,
+			err = srv.brk.RecvEvent(&LiveEvent{
+				Event:      evt,
+				Receiver:   ReceiverGRPCServer,
+				ReceivedAt: time.Now(),
 			})
 			if err != nil {
 				log.Debug(err)

diff --git a/components/broker/engine/http_client.go b/components/broker/engine/http_client.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"net/http"
 	"os"
+	"time"
 
 	"github.com/google/uuid"
 	"github.com/xigxog/kubefox/libs/core/kubefox"
@@ -37,7 +38,7 @@ func NewHTTPClient(brk Broker) *HTTPClient {
 	}
 }
 
-func (c *HTTPClient) SendEvent(req ReceivedEvent) error {
+func (c *HTTPClient) SendEvent(req *LiveEvent) error {
 	ctx, cancel := context.WithTimeout(context.Background(), req.TTL())
 	defer cancel()
 
@@ -51,7 +52,7 @@ func (c *HTTPClient) SendEvent(req ReceivedEvent) error {
 		return err
 	}
 
-	resp := kubefox.StartResp(kubefox.EventOpts{
+	resp := kubefox.NewResp(kubefox.EventOpts{
 		Parent: req.Event,
 		Source: c.comp,
 		Target: req.Source,
@@ -60,10 +61,11 @@ func (c *HTTPClient) SendEvent(req ReceivedEvent) error {
 		return err
 	}
 
-	rEvt := &ReceivedEvent{
-		ActiveEvent:  resp,
-		Subscription: req.Subscription,
+	rEvt := &LiveEvent{
+		Event:        resp,
 		Receiver:     ReceiverHTTPClient,
+		ReceivedAt:   time.Now(),
+		Subscription: req.Subscription,
 	}
 	if err := c.brk.RecvEvent(rEvt); err != nil {
 		return err

diff --git a/components/broker/engine/http_server.go b/components/broker/engine/http_server.go
@@ -136,7 +136,7 @@ func (srv *HTTPServer) ServeHTTP(resWriter http.ResponseWriter, httpReq *http.Re
 
 	resWriter.Header().Set(kubefox.HeaderAdapter, srv.comp.Key())
 
-	req := kubefox.StartReq(kubefox.EventOpts{
+	req := kubefox.NewReq(kubefox.EventOpts{
 		Source: srv.comp,
 	})
 	req.Ttl = config.EventTTL.Microseconds()
@@ -146,7 +146,7 @@ func (srv *HTTPServer) ServeHTTP(resWriter http.ResponseWriter, httpReq *http.Re
 		return
 	}
 
-	log := srv.log.WithEvent(req.Event)
+	log := srv.log.WithEvent(req)
 	log.Debug("received request")
 
 	srv.mutex.Lock()
@@ -160,10 +160,11 @@ func (srv *HTTPServer) ServeHTTP(resWriter http.ResponseWriter, httpReq *http.Re
 		srv.mutex.Unlock()
 	}()
 
-	rEvt := &ReceivedEvent{
-		ActiveEvent: req,
-		Receiver:    ReceiverHTTPServer,
-		ErrCh:       make(chan error),
+	rEvt := &LiveEvent{
+		Event:      req,
+		Receiver:   ReceiverHTTPServer,
+		ReceivedAt: time.Now(),
+		ErrCh:      make(chan error),
 	}
 	if err := srv.brk.RecvEvent(rEvt); err != nil {
 		writeError(resWriter, context.Cause(ctx), log)
@@ -219,7 +220,7 @@ func (srv *HTTPServer) Subscription() Subscription {
 	return srv.sub
 }
 
-func (srv *HTTPServer) sendEvent(mEvt *kubefox.MatchedEvent) error {
+func (srv *HTTPServer) sendEvent(mEvt *LiveEvent) error {
 	resp := mEvt.Event
 	srv.mutex.Lock()
 	respCh, found := srv.reqMap[resp.ParentId]