Remove reflection need for notification log

Author: boekkooi-fresh

driver/sql/postgres/projector_aggregate.go

@@ -162,7 +162,8 @@ func (a *AggregateProjector) processNotification(
 	// Resolve the action to take based on the error that occurred
 	logFields := func(e goengine.LoggerEntry) {
 		e.Error(err)
-		e.Any("notification", notification)
+		e.Int64("notification.no", notification.No)
+		e.String("notification.aggregate_id", notification.AggregateID)
 	}
 	switch resolveErrorAction(a.projectionErrorHandler, notification, err) {
 	case errorFail:
@@ -231,13 +232,15 @@ func (a *AggregateProjector) triggerOutOfSyncProjections(ctx context.Context, qu
 		if err := queue(ctx, notification); err != nil {
 			a.logger.Error("failed to queue notification", func(e goengine.LoggerEntry) {
 				e.Error(err)
-				e.Any("notification", notification)
+				e.Int64("notification.no", notification.No)
+				e.String("notification.aggregate_id", notification.AggregateID)
 			})
 			return err
 		}
 
 		a.logger.Debug("send catchup", func(e goengine.LoggerEntry) {
-			e.Any("notification", notification)
+			e.Int64("notification.no", notification.No)
+			e.String("notification.aggregate_id", notification.AggregateID)
 		})
 	}
 

driver/sql/postgres/projector_aggregate_storage.go

@@ -130,7 +130,8 @@ func (a *aggregateProjectionStorage) PersistState(conn driverSQL.Execer, notific
 	}
 
 	a.logger.Debug("updated projection state", func(e goengine.LoggerEntry) {
-		e.Any("notification", notification)
+		e.Int64("notification.no", notification.No)
+		e.String("notification.aggregate_id", notification.AggregateID)
 		e.Any("state", state)
 	})
 	return nil
@@ -150,7 +151,8 @@ func (a *aggregateProjectionStorage) Acquire(
 	notification *driverSQL.ProjectionNotification,
 ) (func(), *driverSQL.ProjectionRawState, error) {
 	logFields := func(e goengine.LoggerEntry) {
-		e.Any("notification", notification)
+		e.Int64("notification.no", notification.No)
+		e.String("notification.aggregate_id", notification.AggregateID)
 	}
 	aggregateID := notification.AggregateID
 

driver/sql/postgres/projector_stream.go

@@ -148,7 +148,8 @@ func (s *StreamProjector) processNotification(
 		// Resolve the action to take based on the error that occurred
 		logFields := func(e goengine.LoggerEntry) {
 			e.Error(err)
-			e.Any("notification", notification)
+			e.Int64("notification.no", notification.No)
+			e.String("notification.aggregate_id", notification.AggregateID)
 		}
 		switch resolveErrorAction(s.projectionErrorHandler, notification, err) {
 		case errorRetry:

driver/sql/postgres/projector_stream_storage.go

@@ -108,7 +108,8 @@ func (s *streamProjectionStorage) PersistState(conn driverSQL.Execer, notificati
 		return err
 	}
 	s.logger.Debug("updated projection state", func(e goengine.LoggerEntry) {
-		e.Any("notification", notification)
+		e.Int64("notification.no", notification.No)
+		e.String("notification.aggregate_id", notification.AggregateID)
 		e.Any("state", state)
 	})
 
@@ -121,7 +122,8 @@ func (s *streamProjectionStorage) Acquire(
 	notification *driverSQL.ProjectionNotification,
 ) (func(), *driverSQL.ProjectionRawState, error) {
 	logFields := func(e goengine.LoggerEntry) {
-		e.Any("notification", notification)
+		e.Int64("notification.no", notification.No)
+		e.String("notification.aggregate_id", notification.AggregateID)
 	}
 
 	var res *sql.Row

extension/logrus/logrus.go

@@ -95,6 +95,10 @@ func (e entry) Int(k string, v int) {
 	e[k] = v
 }
 
+func (e entry) Int64(k string, v int64) {
+	e[k] = v
+}
+
 func (e entry) String(k, v string) {
 	e[k] = v
 }

extension/pq/listener.go

@@ -153,7 +153,8 @@ func (s *Listener) unmarshalNotification(n *pq.Notification) *sql.ProjectionNoti
 
 	s.logger.Debug("received notification", func(e goengine.LoggerEntry) {
 		e.Any("pq_notification", n)
-		e.Any("notification", notification)
+		e.Int64("notification.no", notification.No)
+		e.String("notification.aggregate_id", notification.AggregateID)
 	})
 
 	return notification

extension/zap/zap.go

@@ -65,6 +65,10 @@ func (e *entry) Int(k string, v int) {
 	e.fields = append(e.fields, zap.Int(k, v))
 }
 
+func (e *entry) Int64(k string, v int64) {
+	e.fields = append(e.fields, zap.Int64(k, v))
+}
+
 func (e *entry) String(k, v string) {
 	e.fields = append(e.fields, zap.String(k, v))
 }

logger.go

@@ -15,6 +15,7 @@ type (
 	// This entry can be enhanced with more date.
 	LoggerEntry interface {
 		Int(k string, v int)
+		Int64(s string, v int64)
 		String(k, v string)
 		Error(err error)
 		Any(k string, v interface{})