Repository: trustwallet/blockatlas
Pull request #943 (2020-03-01 02:57:44)

Update: consumer logic, nginx, compose Add: update event, Ack


Patch:
diff --git a/cmd/observer_subscriber/main.go b/cmd/observer_subscriber/main.go
index 7e32970..4b93f91 100644
--- a/cmd/observer_subscriber/main.go
+++ b/cmd/observer_subscriber/main.go
@@ -7,7 +7,6 @@ import (
 	"github.com/trustwallet/blockatlas/internal"
 	"github.com/trustwallet/blockatlas/mq"
 	"github.com/trustwallet/blockatlas/pkg/logger"
-	"github.com/trustwallet/blockatlas/platform"
 	"github.com/trustwallet/blockatlas/services/subscription"
 	"github.com/trustwallet/blockatlas/storage"
 )
@@ -35,7 +34,6 @@ func init() {
 		logger.Fatal("Failed to init Rabbit MQ", logger.Params{"uri": uri})
 	}
 
-	platform.Init(viper.GetString("platform"))
 }
 
 func main() {
diff --git a/docker-compose.yml b/docker-compose.yml
index 915d746..cae42de 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -11,18 +11,6 @@ services:
       - observer_api
       - platform_api
       - swagger_api
- 
-  observer_api:
-    container_name: observer_api
-    build:
-      context: .
-      args:
-        - SERVICE=observer_api
-    ports:
-      - 8422:8422
-    links:
-      - redis
-    restart: on-failure
 
   platform_api:
     container_name: platform_api
@@ -31,7 +19,7 @@ services:
       args:
         - SERVICE=platform_api
     ports:
-    - 8420:8420
+      - 8420:8420
 
   swagger_api:
     container_name: swagger_api
@@ -42,11 +30,22 @@ services:
     ports:
       - 8423:8423
 
-  platform_observer:
+  observer_worker:
+    build:
+      context: .
+      args:
+        - SERVICE=observer_worker
+    links:
+      - redis
+      - rabbit
+    restart: on-failure
+
+  observer_subscriber:
+    container_name: observer_subscriber
     build:
       context: .
       args:
-        - SERVICE=platform_observer
+        - SERVICE=observer_subscriber
     links:
       - redis
       - rabbit
@@ -67,4 +66,4 @@ services:
     container_name: redis
     image: redis
     ports:
-    - 6379:6379
+      - 6379:6379
diff --git a/nginx.conf b/nginx.conf
index 77b5be9..8fd11f9 100644
--- a/nginx.conf
+++ b/nginx.conf
@@ -5,11 +5,7 @@ server {
         proxy_pass http://swagger_api:8423;
     }
 
-    location ~* /observer.* {
-        proxy_pass http://observer_api:8422;
-    }
-
     location / {
         proxy_pass http://platform_api:8420;
     }
-}
\ No newline at end of file
+}
diff --git a/pkg/blockatlas/observer.go b/pkg/blockatlas/observer.go
index dc5909c..7b7e79e 100644
--- a/pkg/blockatlas/observer.go
+++ b/pkg/blockatlas/observer.go
@@ -7,9 +7,10 @@ type Subscriptions map[string][]string
 type SubscriptionOperation string
 
 type SubscriptionEvent struct {
-	Subscriptions Subscriptions         `json:"subscriptions"`
-	GUID          string                `json:"guid"`
-	Operation     SubscriptionOperation `json:"operation"`
+	NewSubscriptions Subscriptions         `json:"new_subscriptions"`
+	OldSubscriptions Subscriptions         `json:"old_subscriptions"`
+	GUID             string                `json:"guid"`
+	Operation        SubscriptionOperation `json:"operation"`
 }
 
 type CoinStatus struct {
@@ -22,9 +23,9 @@ type Observer struct {
 	Message string `json:"message"`
 }
 
-func (e *SubscriptionEvent) ParseSubscriptions() []Subscription {
+func (e *SubscriptionEvent) ParseSubscriptions(s Subscriptions) []Subscription {
 	subs := make([]Subscription, 0)
-	for coinStr, perCoin := range e.Subscriptions {
+	for coinStr, perCoin := range s {
 		coin, err := strconv.Atoi(coinStr)
 		if err != nil {
 			continue
diff --git a/services/subscription/subscribtion.go b/services/subscription/subscribtion.go
index 2871511..518875b 100644
--- a/services/subscription/subscribtion.go
+++ b/services/subscription/subscribtion.go
@@ -11,6 +11,7 @@ import (
 const (
 	AddSubscription    blockatlas.SubscriptionOperation = "AddSubscription"
 	DeleteSubscription blockatlas.SubscriptionOperation = "DeleteSubscription"
+	UpdateSubscription blockatlas.SubscriptionOperation = "UpdateSubscription"
 )
 
 func Consume(delivery amqp.Delivery, storage storage.Addresses) {
@@ -19,20 +20,44 @@ func Consume(delivery amqp.Delivery, storage storage.Addresses) {
 	if err != nil {
 		logger.Fatal(err)
 	}
-	subscriptions := event.ParseSubscriptions()
+	newSubscriptions := event.ParseSubscriptions(event.NewSubscriptions)
+	oldSubscriptions := event.ParseSubscriptions(event.OldSubscriptions)
+	params := logger.Params{"operation": event.Operation, "guid": event.GUID, "new_subscriptions_len": len(newSubscriptions), "old_subscriptions_len": len(oldSubscriptions)}
 
 	switch event.Operation {
+	case UpdateSubscription:
+		err := storage.DeleteSubscriptions(oldSubscriptions)
+		if err != nil {
+			logger.Error(err, params)
+		}
+		err = storage.AddSubscriptions(newSubscriptions)
+		if err != nil {
+			logger.Error(err, params)
+		}
+		err = delivery.Ack(false)
+		if err != nil {
+			logger.Error(err, params)
+		}
+		logger.Info("Updated", params)
 	case AddSubscription:
-		err := storage.AddSubscriptions(subscriptions)
+		err = storage.AddSubscriptions(newSubscriptions)
+		if err != nil {
+			logger.Error(err, params)
+		}
+		err = delivery.Ack(false)
 		if err != nil {
-			logger.Fatal(err, logger.Params{"operation": event.Operation, "guid": event.GUID})
+			logger.Error(err, params)
 		}
-		logger.Info("Success", logger.Params{"operation": event.Operation, "guid": event.GUID, "subscriptions_len": len(subscriptions)})
+		logger.Info("Added", params)
 	case DeleteSubscription:
-		err := storage.DeleteSubscriptions(nil)
+		err := storage.DeleteSubscriptions(oldSubscriptions)
+		if err != nil {
+			logger.Error(err, params)
+		}
+		err = delivery.Ack(false)
 		if err != nil {
-			logger.Fatal(err, logger.Params{"operation": event.Operation, "guid": event.GUID})
+			logger.Error(err, params)
 		}
-		logger.Info("Success", logger.Params{"operation": event.Operation, "guid": event.GUID, "subscriptions_len": len(subscriptions)})
+		logger.Info("Deleted", params)
 	}
 }


Test patch:
diff --git a/pkg/blockatlas/observer_test.go b/pkg/blockatlas/observer_test.go
index bd56ef6..1f80e0e 100644
--- a/pkg/blockatlas/observer_test.go
+++ b/pkg/blockatlas/observer_test.go
@@ -16,7 +16,7 @@ func Test_parseSubscriptions(t *testing.T) {
 			name: "guid with 1 coin",
 			subscriptions: SubscriptionEvent{
 				GUID: "http://127.0.0.1:8080",
-				Subscriptions: Subscriptions{
+				OldSubscriptions: Subscriptions{
 					"0": {"xpub6BpYi6J1GZzfY3yY7DbhLLccF3efQa18nQngM3jaehgtNSoEgk6UtPULpC3oK5oA3trczY8Ld34LFw1USMPfGHwTEizdD5QyGcMyuh2UoBA", "xpub6CYwPfnPJLPquufPkb98coSb3mdy1CgaZrWUtYWGJTJ4VWZUbzH9HLGy7nHpP7DG4UdTkYYpirkTWQSP7pWHsrk24Nos5oYNHpfr4BgPVTL"},
 				},
 			},
@@ -35,7 +35,7 @@ func Test_parseSubscriptions(t *testing.T) {
 			name: "guid with 2 coins",
 			subscriptions: SubscriptionEvent{
 				GUID: "http://127.0.0.1:8080",
-				Subscriptions: Subscriptions{
+				OldSubscriptions: Subscriptions{
 					"2": {"zpub6rH4MwgyTmuexAX6HAraks5cKv5BbtmwdLirvnU5845ovUJb4abgjt9DtXK4ZEaToRrNj8dQznuLC6Nka4eMviGMinCVMUxKLpuyddcG9Vc"},
 					"0": {"xpub6BpYi6J1GZzfY3yY7DbhLLccF3efQa18nQngM3jaehgtNSoEgk6UtPULpC3oK5oA3trczY8Ld34LFw1USMPfGHwTEizdD5QyGcMyuh2UoBA", "xpub6CYwPfnPJLPquufPkb98coSb3mdy1CgaZrWUtYWGJTJ4VWZUbzH9HLGy7nHpP7DG4UdTkYYpirkTWQSP7pWHsrk24Nos5oYNHpfr4BgPVTL"},
 				},
@@ -58,7 +58,7 @@ func Test_parseSubscriptions(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			gotSubs := tt.subscriptions.ParseSubscriptions()
+			gotSubs := tt.subscriptions.ParseSubscriptions(tt.subscriptions.OldSubscriptions)
 			sort.Slice(gotSubs, func(i, j int) bool {
 				return gotSubs[i].Coin > gotSubs[j].Coin
 			})