Merge #7

7: Bump ESModel to v4.0.0 r=Jaskaranbir a=Jaskaranbir

bors r+

Co-authored-by: Jaskaranbir <jaskaranbir.dhillon@gmail.com>

.env

@@ -1,3 +1,6 @@
+SERVICE_NAME=agg-userauth-cmd
+KAFKA_LOG_PRODUCER_TOPIC=log.sink
+
 # ===> Kafka
 KAFKA_BROKERS=kafka:9092
 

Gopkg.toml

@@ -38,11 +38,11 @@
 
 [[constraint]]
   name = "github.com/TerrexTech/go-eventspoll"
-  version = "3.0.0"
+  version = "4.0.0"
 
 [[constraint]]
   name = "github.com/TerrexTech/go-eventstore-models"
-  version = "3.0.0"
+  version = "4.0.0"
 
 [[constraint]]
   name = "github.com/TerrexTech/go-kafkautils"

main/main.go

@@ -81,7 +81,7 @@ func main() {
 
 	for {
 		select {
-		case <-eventPoll.RoutinesCtx().Done():
+		case err := <-eventPoll.Wait():
 			err = errors.New("service-context closed")
 			log.Fatalln(err)
 

test/.envp

@@ -1,3 +1,6 @@
+SERVICE_NAME=go-eventstore-query
+KAFKA_LOG_PRODUCER_TOPIC=log.sink
+
 CASSANDRA_HOSTS=cassandra:9042
 CASSANDRA_DATA_CENTERS=datacenter1:1
 CASSANDRA_USERNAME=cassandra

test/.envq

@@ -1,3 +1,6 @@
+SERVICE_NAME=go-eventpersistence
+KAFKA_LOG_PRODUCER_TOPIC=log.sink
+
 CASSANDRA_HOSTS=cassandra:9042
 CASSANDRA_DATA_CENTERS=datacenter1:1
 CASSANDRA_EVENT_META_TABLE=events_meta

test/docker-compose.yaml

@@ -34,12 +34,12 @@ services:
       - 2181:2181
 
   go-eventpersistence:
-    image: terrextech/go-eventpersistence:v2.0.1
+    image: terrextech/go-eventpersistence:v3.0.0
     env_file:
       - ./.envp
 
   go-eventstore-query:
-    image: terrextech/go-eventstore-query:v2.1.0
+    image: terrextech/go-eventstore-query:v3.0.0
     env_file:
       - ./.envq
 

test/user_suite_test.go

@@ -100,16 +100,16 @@ var _ = Describe("UserAuthAggregate", func() {
 		Expect(err).ToNot(HaveOccurred())
 		uid, err := uuuid.NewV4()
 		Expect(err).ToNot(HaveOccurred())
-		timeUUID, err := uuuid.NewV1()
+		UUID, err := uuuid.NewV4()
 		Expect(err).ToNot(HaveOccurred())
 		mockEvent = &model.Event{
-			Action:        "insert",
+			EventAction:   "insert",
 			CorrelationID: cid,
 			AggregateID:   user.AggregateID,
 			Data:          marshalUser,
-			Timestamp:     time.Now(),
+			NanoTime:      time.Now().UnixNano(),
 			UserUUID:      uid,
-			TimeUUID:      timeUUID,
+			UUID:          UUID,
 			Version:       0,
 			YearBucket:    2018,
 		}
@@ -139,11 +139,11 @@ var _ = Describe("UserAuthAggregate", func() {
 				err := json.Unmarshal(msg.Value, kr)
 				Expect(err).ToNot(HaveOccurred())
 
-				if kr.UUID == mockEvent.TimeUUID {
+				if kr.UUID == mockEvent.UUID {
 					Expect(kr.Error).To(BeEmpty())
 					Expect(kr.ErrorCode).To(BeZero())
 					Expect(kr.CorrelationID).To(Equal(mockEvent.CorrelationID))
-					Expect(kr.UUID).To(Equal(mockEvent.TimeUUID))
+					Expect(kr.UUID).To(Equal(mockEvent.UUID))
 
 					user := &user.User{}
 					err = json.Unmarshal(kr.Result, user)
@@ -193,12 +193,12 @@ var _ = Describe("UserAuthAggregate", func() {
 			mockUser.ID = mockID
 
 			Byf("Creating update MockEvent")
-			timeUUID, err := uuuid.NewV1()
+			UUID, err := uuuid.NewV4()
 			Expect(err).ToNot(HaveOccurred())
-			mockEvent.Action = "update"
+			mockEvent.EventAction = "update"
 			mockEvent.Data = marshalUpdate
-			mockEvent.Timestamp = time.Now()
-			mockEvent.TimeUUID = timeUUID
+			mockEvent.NanoTime = time.Now().UnixNano()
+			mockEvent.UUID = UUID
 
 			Byf("Producing MockEvent")
 			p, err := kafka.NewProducer(&kafka.ProducerConfig{
@@ -222,11 +222,11 @@ var _ = Describe("UserAuthAggregate", func() {
 				err := json.Unmarshal(msg.Value, kr)
 				Expect(err).ToNot(HaveOccurred())
 
-				if kr.UUID == mockEvent.TimeUUID {
+				if kr.UUID == mockEvent.UUID {
 					Expect(kr.Error).To(BeEmpty())
 					Expect(kr.ErrorCode).To(BeZero())
 					Expect(kr.CorrelationID).To(Equal(mockEvent.CorrelationID))
-					Expect(kr.UUID).To(Equal(mockEvent.TimeUUID))
+					Expect(kr.UUID).To(Equal(mockEvent.UUID))
 
 					result := map[string]int{}
 					err = json.Unmarshal(kr.Result, &result)
@@ -265,12 +265,12 @@ var _ = Describe("UserAuthAggregate", func() {
 			Expect(err).ToNot(HaveOccurred())
 
 			Byf("Creating delete MockEvent")
-			timeUUID, err := uuuid.NewV1()
+			UUID, err := uuuid.NewV4()
 			Expect(err).ToNot(HaveOccurred())
-			mockEvent.Action = "delete"
+			mockEvent.EventAction = "delete"
 			mockEvent.Data = marshalDelete
-			mockEvent.Timestamp = time.Now()
-			mockEvent.TimeUUID = timeUUID
+			mockEvent.NanoTime = time.Now().UnixNano()
+			mockEvent.UUID = UUID
 
 			Byf("Producing MockEvent")
 			p, err := kafka.NewProducer(&kafka.ProducerConfig{
@@ -294,11 +294,11 @@ var _ = Describe("UserAuthAggregate", func() {
 				err := json.Unmarshal(msg.Value, kr)
 				Expect(err).ToNot(HaveOccurred())
 
-				if kr.UUID == mockEvent.TimeUUID {
+				if kr.UUID == mockEvent.UUID {
 					Expect(kr.Error).To(BeEmpty())
 					Expect(kr.ErrorCode).To(BeZero())
 					Expect(kr.CorrelationID).To(Equal(mockEvent.CorrelationID))
-					Expect(kr.UUID).To(Equal(mockEvent.TimeUUID))
+					Expect(kr.UUID).To(Equal(mockEvent.UUID))
 
 					result := map[string]int{}
 					err = json.Unmarshal(kr.Result, &result)

user/delete.go

@@ -26,7 +26,9 @@ func Delete(collection *mongo.Collection, event *model.Event) *model.KafkaRespon
 			CorrelationID: event.CorrelationID,
 			Error:         err.Error(),
 			ErrorCode:     InternalError,
-			UUID:          event.TimeUUID,
+			EventAction:   event.EventAction,
+			ServiceAction: event.ServiceAction,
+			UUID:          event.UUID,
 		}
 	}
 
@@ -39,7 +41,9 @@ func Delete(collection *mongo.Collection, event *model.Event) *model.KafkaRespon
 			CorrelationID: event.CorrelationID,
 			Error:         err.Error(),
 			ErrorCode:     InternalError,
-			UUID:          event.TimeUUID,
+			EventAction:   event.EventAction,
+			ServiceAction: event.ServiceAction,
+			UUID:          event.UUID,
 		}
 	}
 
@@ -52,7 +56,9 @@ func Delete(collection *mongo.Collection, event *model.Event) *model.KafkaRespon
 			CorrelationID: event.CorrelationID,
 			Error:         err.Error(),
 			ErrorCode:     DatabaseError,
-			UUID:          event.TimeUUID,
+			EventAction:   event.EventAction,
+			ServiceAction: event.ServiceAction,
+			UUID:          event.UUID,
 		}
 	}
 
@@ -66,14 +72,18 @@ func Delete(collection *mongo.Collection, event *model.Event) *model.KafkaRespon
 			CorrelationID: event.CorrelationID,
 			Error:         err.Error(),
 			ErrorCode:     InternalError,
-			UUID:          event.TimeUUID,
+			EventAction:   event.EventAction,
+			ServiceAction: event.ServiceAction,
+			UUID:          event.UUID,
 		}
 	}
 
 	return &model.KafkaResponse{
 		AggregateID:   event.AggregateID,
 		CorrelationID: event.CorrelationID,
+		EventAction:   event.EventAction,
 		Result:        resultMarshal,
-		UUID:          event.TimeUUID,
+		ServiceAction: event.ServiceAction,
+		UUID:          event.UUID,
 	}
 }