Repository: trustwallet/blockatlas
Pull request #958 (2020-03-11 03:54:50)

MQ prefetch && refactor


Patch:
diff --git a/Makefile b/Makefile
index b05a97f..d3e4582 100644
--- a/Makefile
+++ b/Makefile
@@ -193,7 +193,7 @@ go-functional:
 
 go-integration:
 	@echo "  >  Running integration tests"
-	GOBIN=$(GOBIN) TEST_CONFIG=$(CONFIG_FILE) go test -race -tags=integration -v ./tests/integration
+	GOBIN=$(GOBIN) TEST_CONFIG=$(CONFIG_FILE) go test -race -tags=integration -v ./tests/integration ./tests/docker_test
 
 go-fmt:
 	@echo "  >  Format all go files"
diff --git a/cmd/observer_subscriber/main.go b/cmd/observer_subscriber/main.go
index 8713bc4..b6b414b 100644
--- a/cmd/observer_subscriber/main.go
+++ b/cmd/observer_subscriber/main.go
@@ -1,7 +1,6 @@
 package main
 
 import (
-	"github.com/gin-gonic/gin"
 	"github.com/spf13/viper"
 	_ "github.com/trustwallet/blockatlas/docs"
 	"github.com/trustwallet/blockatlas/internal"
@@ -16,29 +15,26 @@ const (
 )
 
 var (
-	port, confPath string
-	cache          *storage.Storage
-	sg             *gin.HandlerFunc
+	confPath string
+	cache    *storage.Storage
 )
 
 func init() {
-	_, confPath, _, cache = internal.InitAPIWithRedis("", defaultConfigPath)
-
-	uri := viper.GetString("observer.rabbitmq.uri")
-	err := mq.Init(uri)
-	if err != nil {
-		logger.Fatal("Failed to init Rabbit MQ", logger.Params{"uri": uri})
-	}
-	err = mq.Transactions.Declare()
-	if err != nil {
-		logger.Fatal("Failed to init Rabbit MQ", logger.Params{"uri": uri})
-	}
+	_, confPath := internal.ParseArgs("", defaultConfigPath)
 
+	internal.InitConfig(confPath)
+	logger.InitLogger()
+
+	redisHost := viper.GetString("storage.redis")
+	mqHost := viper.GetString("observer.rabbitmq.uri")
+	prefetchCount := viper.GetInt("observer.rabbitmq.consumer.prefetch_count")
+
+	cache = internal.InitRedis(redisHost)
+	internal.InitRabbitMQ(mqHost, prefetchCount)
 }
 
 func main() {
 	defer mq.Close()
 	mq.Subscriptions.RunConsumer(subscription.Consume, cache)
 	<-make(chan struct{})
-	mq.Close()
 }
diff --git a/cmd/observer_worker/main.go b/cmd/observer_worker/main.go
index e126abe..5e2c3d1 100644
--- a/cmd/observer_worker/main.go
+++ b/cmd/observer_worker/main.go
@@ -22,22 +22,24 @@ var (
 )
 
 func init() {
-	_, confPath, _, cache = internal.InitAPIWithRedis("", defaultConfigPath)
+	_, confPath := internal.ParseArgs("", defaultConfigPath)
 
-	uri := viper.GetString("observer.rabbitmq.uri")
-	err := mq.Init(uri)
-	if err != nil {
-		logger.Fatal("Failed to init Rabbit MQ", logger.Params{"uri": uri})
-	}
-	err = mq.Transactions.Declare()
-	if err != nil {
-		logger.Fatal("Failed to init Rabbit MQ", logger.Params{"uri": uri})
-	}
+	internal.InitConfig(confPath)
+	logger.InitLogger()
 
-	platform.Init(viper.GetString("platform"))
+	redisHost := viper.GetString("storage.redis")
+	mqHost := viper.GetString("observer.rabbitmq.uri")
+	prefetchCount := viper.GetInt("observer.rabbitmq.consumer.prefetch_count")
+	platformHandle := viper.GetString("platform")
+
+	cache = internal.InitRedis(redisHost)
+	internal.InitRabbitMQ(mqHost, prefetchCount)
+	platform.Init(platformHandle)
 }
 
 func main() {
+	defer mq.Close()
+
 	if len(platform.BlockAPIs) == 0 {
 		logger.Fatal("No APIs to observe")
 	}
@@ -96,6 +98,5 @@ func main() {
 	}
 
 	wg.Wait()
-	mq.Close()
 	logger.Info("Exiting cleanly")
 }
diff --git a/cmd/platform_api/main.go b/cmd/platform_api/main.go
index 26ac439..4e293b0 100644
--- a/cmd/platform_api/main.go
+++ b/cmd/platform_api/main.go
@@ -1,12 +1,13 @@
 package main
 
 import (
+	sentrygin "github.com/getsentry/sentry-go/gin"
 	"github.com/gin-gonic/gin"
 	"github.com/spf13/viper"
 	"github.com/trustwallet/blockatlas/api"
 	_ "github.com/trustwallet/blockatlas/docs"
 	"github.com/trustwallet/blockatlas/internal"
-	"github.com/trustwallet/blockatlas/pkg/ginutils"
+	"github.com/trustwallet/blockatlas/pkg/logger"
 	"github.com/trustwallet/blockatlas/platform"
 )
 
@@ -17,22 +18,23 @@ const (
 
 var (
 	port, confPath string
-	sg             *gin.HandlerFunc
+	engine         *gin.Engine
 )
 
 func init() {
-	port, confPath, sg = internal.InitAPI(defaultPort, defaultConfigPath)
+	port, confPath = internal.ParseArgs(defaultPort, defaultConfigPath)
+
+	internal.InitConfig(confPath)
+	logger.InitLogger()
+	tmp := sentrygin.New(sentrygin.Options{})
+	sg := &tmp
+
+	engine = internal.InitEngine(sg, viper.GetString("gin.mode"))
+
 	platform.Init(viper.GetString("platform"))
 }
 
 func main() {
-	gin.SetMode(viper.GetString("gin.mode"))
-
-	engine := gin.New()
-	engine.Use(ginutils.CheckReverseProxy, *sg)
-	engine.Use(ginutils.CORSMiddleware())
-	engine.Use(gin.Logger())
-
 	api.SetupPlatformAPI(engine)
 	internal.SetupGracefulShutdown(port, engine)
 }
diff --git a/cmd/swagger_api/main.go b/cmd/swagger_api/main.go
index 2d2c4b4..ceb9c0d 100644
--- a/cmd/swagger_api/main.go
+++ b/cmd/swagger_api/main.go
@@ -1,14 +1,15 @@
 package main
 
 import (
+	sentrygin "github.com/getsentry/sentry-go/gin"
 	"github.com/gin-gonic/gin"
 	"github.com/spf13/viper"
 	ginSwagger "github.com/swaggo/gin-swagger"
 	"github.com/swaggo/gin-swagger/swaggerFiles"
-	"github.com/trustwallet/blockatlas/api"
 	_ "github.com/trustwallet/blockatlas/docs"
 	"github.com/trustwallet/blockatlas/internal"
-	"github.com/trustwallet/blockatlas/pkg/ginutils"
+	"github.com/trustwallet/blockatlas/pkg/logger"
+	"net/http"
 )
 
 const (
@@ -18,23 +19,22 @@ const (
 
 var (
 	port, confPath string
-	sg             *gin.HandlerFunc
+	engine         *gin.Engine
 )
 
 func init() {
-	port, confPath, sg = internal.InitAPI(defaultPort, defaultConfigPath)
+	port, confPath = internal.ParseArgs(defaultPort, defaultConfigPath)
+	tmp := sentrygin.New(sentrygin.Options{})
+	sg := &tmp
+	internal.InitConfig(confPath)
+	logger.InitLogger()
+	engine = internal.InitEngine(sg, viper.GetString("gin.mode"))
 }
 
 func main() {
-	gin.SetMode(viper.GetString("gin.mode"))
-	engine := gin.New()
-
-	engine.Use(ginutils.CheckReverseProxy, *sg)
-	engine.Use(ginutils.CORSMiddleware())
-
-	engine.GET("/", api.GetRoot)
-	engine.GET("/status", api.GetStatus)
+	logger.Info("Loading Swagger API")
+	engine.GET("/", func(c *gin.Context) {
+		c.Redirect(http.StatusMovedPermanently, "swagger/index.html")
+	})
 	engine.GET("swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler))
-
-	internal.SetupGracefulShutdown(port, engine)
 }
diff --git a/config.yml b/config.yml
index 65444ad..da45859 100644
--- a/config.yml
+++ b/config.yml
@@ -29,10 +29,12 @@ observer:
     min: 3s
     max: 30s
   rabbitmq:
-    uri: amqp://user:bitnami@rabbit:5672
+    uri: amqp://rabbit:5672
+    consumer:
+      prefetch_count: 10
 
 storage:
-  redis: redis://redis:6379
+  redis: redis://localhost:6379
 
 # [BNB] Binance DEX: https://wallet.binance.org
 #       Binance Chain: https://explorer.binance.org
diff --git a/docker-compose.yml b/docker-compose.yml
index cae42de..9d85621 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -53,7 +53,7 @@ services:
 
   rabbit:
     container_name: rabbit
-    image: bitnami/rabbitmq
+    image: rabbitmq
     ports:
       - 5672:5672
     healthcheck:
diff --git a/go.mod b/go.mod
index de0bb6a..b5168b5 100644
--- a/go.mod
+++ b/go.mod
@@ -3,21 +3,32 @@ module github.com/trustwallet/blockatlas
 go 1.12
 
 require (
+	github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78 // indirect
+	github.com/Microsoft/go-winio v0.4.14 // indirect
+	github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 // indirect
 	github.com/Pantani/httpexpect v2.0.0+incompatible
 	github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751
 	github.com/btcsuite/btcutil v1.0.1
 	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/chenjiandongx/ginprom v0.0.0-20191227144730-e11ebf56bc05
+	github.com/containerd/continuity v0.0.0-20200228182428-0f16d7a0959c // indirect
 	github.com/deckarep/golang-set v1.7.1
+	github.com/docker/go-connections v0.4.0 // indirect
+	github.com/docker/go-units v0.4.0 // indirect
 	github.com/ethereum/go-ethereum v1.9.10
 	github.com/getsentry/sentry-go v0.4.0
 	github.com/gin-gonic/gin v1.5.0
 	github.com/go-redis/redis v6.15.6+incompatible
 	github.com/hewigovens/go-coincodec v1.0.4
 	github.com/jinzhu/gorm v1.9.12
+	github.com/konsorten/go-windows-terminal-sequences v1.0.2 // indirect
 	github.com/mitchellh/mapstructure v1.1.2
 	github.com/mr-tron/base58 v1.1.3
+	github.com/opencontainers/image-spec v1.0.1 // indirect
+	github.com/opencontainers/runc v0.1.1 // indirect
+	github.com/ory/dockertest v3.3.5+incompatible
 	github.com/patrickmn/go-cache v2.1.0+incompatible
+	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.4.0
 	github.com/robfig/cron/v3 v3.0.1
 	github.com/shopspring/decimal v0.0.0-20200105231215-408a2507e114
@@ -30,5 +41,7 @@ require (
 	github.com/swaggo/swag v1.6.5
 	github.com/wealdtech/go-ens/v3 v3.2.0
 	golang.org/x/crypto v0.0.0-20200128174031-69ecbb4d6d5d
+	golang.org/x/net v0.0.0-20200301022130-244492dfa37a // indirect
+	golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527 // indirect
 	gopkg.in/yaml.v2 v2.2.8
 )
diff --git a/go.sum b/go.sum
index 193793c..849be02 100644
--- a/go.sum
+++ b/go.sum
@@ -1,8 +1,11 @@
+bazil.org/fuse v0.0.0-20160811212531-371fbbdaa898/go.mod h1:Xbm+BRKSBEpa4q4hTSxohYNQpsxXPbPry4JJWOB3LB8=
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 github.com/AndreasBriese/bbloom v0.0.0-20190306092124-e2d15f34fcf9/go.mod h1:bOvUY6CB00SOBii9/FifXqc0awNKxLFCL/+pkDPuyl8=
 github.com/Azure/azure-pipeline-go v0.2.1/go.mod h1:UGSo8XybXnIGZ3epmeBw7Jdz+HiUVpqIlpz/HKHylF4=
 github.com/Azure/azure-pipeline-go v0.2.2/go.mod h1:4rQ/NZncSvGqNkkOsNpOU1tgoNuIlp9AfUH5G1tvCHc=
 github.com/Azure/azure-storage-blob-go v0.7.0/go.mod h1:f9YQKtsG1nMisotuTPpO0tjNuEjKRYAcJU8/ydDI++4=
+github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78 h1:w+iIsaOQNcT7OZ575w+acHgRric5iCyQh+xv+KJ4HB8=
+github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8=
 github.com/Azure/go-autorest/autorest v0.9.0/go.mod h1:xyHB1BMZT0cuDHU7I0+g046+BFDTQ8rEZB0s4Yfa6bI=
 github.com/Azure/go-autorest/autorest/adal v0.5.0/go.mod h1:8Z9fGy2MpX0PvDjB1pEgQTmVqjGhiHBW7RJJEciWzS0=
 github.com/Azure/go-autorest/autorest/adal v0.8.0/go.mod h1:Z6vX6WXXuyieHAXwMj0S6HY6e6wcHn37qQMBQlvY3lc=
@@ -20,6 +23,10 @@ github.com/Joker/hpp v1.0.0/go.mod h1:8x5n+M1Hp5hC0g8okX3sR3vFQwynaX/UgSOM9MeBKz
 github.com/Joker/jade v1.0.1-0.20190614124447-d475f43051e7/go.mod h1:6E6s8o2AE4KhCrqr6GRJjdC/gNfTdxkIXvuGZZda2VM=
 github.com/KyleBanks/depth v1.2.1 h1:5h8fQADFrWtarTdtDudMmGsC7GPbOAu6RVB3ffsVFHc=
 github.com/KyleBanks/depth v1.2.1/go.mod h1:jzSb9d0L43HxTQfT+oSA1EEp2q+ne2uh6XgeJcm8brE=
+github.com/Microsoft/go-winio v0.4.14 h1:+hMXMk01us9KgxGb7ftKQt2Xpf5hH/yky+TDA+qxleU=
+github.com/Microsoft/go-winio v0.4.14/go.mod h1:qXqCSQ3Xa7+6tgxaGTIe4Kpcdsi+P8jBhyzoq1bpyYA=
+github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 h1:TngWCqHvy9oXAN6lEVMRuU21PR1EtLVZJmdB18Gu3Rw=
+github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5/go.mod h1:lmUJ/7eu/Q8D7ML55dXQrVaamCz2vxCfdQBasLZfHKk=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/OneOfOne/xxhash v1.2.5/go.mod h1:eZbhyaAYD41SGSSsnmcpxVoRiQ/MPUTjUdIIOT9Um7Q=
 github.com/Pantani/httpexpect v2.0.0+incompatible h1:P658jo0d3gej0D67KBcSUgpZY76CH1TdxvOfQAmyNu0=
@@ -84,6 +91,8 @@ github.com/chenjiandongx/ginprom v0.0.0-20191227144730-e11ebf56bc05/go.mod h1:lI
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cloudflare/cloudflare-go v0.10.2-0.20190916151808-a80f83b9add9/go.mod h1:1MxXX1Ux4x6mqPmjkUgTP1CdXIBXKX7T+Jk9Gxrmx+U=
 github.com/codegangsta/inject v0.0.0-20150114235600-33e0aa1cb7c0/go.mod h1:4Zcjuz89kmFXt9morQgcfYZAYZ5n8WHjt81YYWIwtTM=
+github.com/containerd/continuity v0.0.0-20200228182428-0f16d7a0959c h1:8ahmSVELW1wghbjerVAyuEYD5+Dio66RYvSS0iGfL1M=
+github.com/containerd/continuity v0.0.0-20200228182428-0f16d7a0959c/go.mod h1:Dq467ZllaHgAtVp4p1xUQWBrFXR9s/wyoTpG8zOJGkY=
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8NzMklzPG4d5KIOhIy30Tk=
@@ -109,6 +118,11 @@ github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUn
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
 github.com/docker/docker v1.4.2-0.20180625184442-8e610b2b55bf/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/docker v1.13.1/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
+github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKohAFqRJQ=
+github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec=
+github.com/docker/go-units v0.4.0 h1:3uh0PgVws3nIA0Q+MwDC8yjEPf9zjRfZZWXZYDct3Tw=
+github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
+github.com/dustin/go-humanize v0.0.0-20171111073723-bb3d318650d4/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/edsrzf/mmap-go v0.0.0-20160512033002-935e0e8a636c/go.mod h1:YO35OhQPt3KJa3ryjFM5Bs14WD66h8eGKpfaBNrHW5M=
 github.com/edsrzf/mmap-go v1.0.0 h1:CEBF7HpRnUCSJgGUb5h1Gm7e3VkmVDrR8lvWVLtrOFw=
@@ -280,7 +294,10 @@ github.com/klauspost/compress v1.8.2/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0
 github.com/klauspost/compress v1.9.0 h1:GhthINjveNZAdFUD8QoQYfjxnOONZgztK/Yr6M23UTY=
 github.com/klauspost/compress v1.9.0/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
 github.com/klauspost/cpuid v1.2.1/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
+github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGiHgQ4OO8tzTaLawm8vnODuwDk=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
+github.com/konsorten/go-windows-terminal-sequences v1.0.2 h1:DB17ag19krx9CFsz4o3enTrPXyIXCl+2iCXH/aMAp9s=
+github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
@@ -354,10 +371,20 @@ github.com/olekukonko/tablewriter v0.0.2-0.20190409134802-7e037d187b0c h1:1RHs3t
 github.com/olekukonko/tablewriter v0.0.2-0.20190409134802-7e037d187b0c/go.mod h1:vsDQFd/mU46D+Z4whnwzcISnGGzXWMclvtLoiIKAKIo=
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.10.3/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
+github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
 github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
+github.com/opencontainers/go-digest v1.0.0-rc1 h1:WzifXhOVOEOuFYOJAW6aQqW0TooG2iki3E3Ii+WN7gQ=
+github.com/opencontainers/go-digest v1.0.0-rc1/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s=
+github.com/opencontainers/image-spec v1.0.1 h1:JMemWkRwHx4Zj+fVxWoMCFm/8sYGGrUVojFA6h/TRcI=
+github.com/opencontainers/image-spec v1.0.1/go.mod h1:BtxoFyWECRxE4U/7sNtV5W15zMzWCbyJoFRP3s7yZA0=
+github.com/opencontainers/runc v0.1.1 h1:GlxAyO6x8rfZYN9Tt0Kti5a/cP41iuiO2yYT0IJGY8Y=
+github.com/opencontainers/runc v0.1.1/go.mod h1:qT5XzbpPznkRYVz/mWwUaVBUv2rmF59PVA73FjuZG0U=
 github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
+github.com/ory/dockertest v3.3.5+incompatible h1:iLLK6SQwIhcbrG783Dghaaa3WPzGc+4Emza6EbVUUGA=
+github.com/ory/dockertest v3.3.5+incompatible/go.mod h1:1vX4m9wsvi00u5bseYwXaSnhNrne+V0E6LAcBILJdPs=
 github.com/oschwald/maxminddb-golang v1.3.1/go.mod h1:3jhIUymTJ5VREKyIhWm66LJiQt04F0UCDdodShpjWsY=
 github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
 github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
@@ -369,8 +396,11 @@ github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/9
 github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7/go.mod h1:CRroGNssyjTd/qIG2FyxByd2S8JEAZXBl4qUrZf8GS0=
 github.com/pingcap/errors v0.11.4/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pkg/errors v0.8.1-0.20171018195549-f15c970de5b7/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
@@ -416,7 +446,9 @@ github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAm
 github.com/shopspring/decimal v0.0.0-20200105231215-408a2507e114 h1:Pm6R878vxWWWR+Sa3ppsLce/Zq+JNTs6aVvRu13jv9A=
 github.com/shopspring/decimal v0.0.0-20200105231215-408a2507e114/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
 github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
+github.com/sirupsen/logrus v1.0.4-0.20170822132746-89742aefa4b2/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
+github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMBDgk/93Q=
 github.com/sirupsen/logrus v1.4.2 h1:SPIRibHv4MatM3XXNO2BJeFLZwZ2LvZgfQ5+UNI2im4=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d h1:zE9ykElWQ6/NYmHa3jpm/yHnI4xSofP+UP6SpjHcSeM=
@@ -432,9 +464,11 @@ github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B
 github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
 github.com/spf13/cast v1.3.1 h1:nFm6S0SMdyzrzcmThSipiEubIDy8WEXKNZ0UOgiRpng=
 github.com/spf13/cast v1.3.1/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
+github.com/spf13/cobra v0.0.2-0.20171109065643-2da4a54c5cee/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ=
 github.com/spf13/cobra v0.0.5/go.mod h1:3K3wKZymM7VvHMDS9+Akkh4K60UwM26emMESw8tLCHU=
 github.com/spf13/jwalterweatherman v1.0.0 h1:XHEdyB+EcvlqZamSM4ZOMGlc93t6AcsBEu9Gc1vn7yk=
 github.com/spf13/jwalterweatherman v1.0.0/go.mod h1:cQK4TGJAtQXfYWX+Ddv3mKDzgVb68N+wFjFa4jdeBTo=
+github.com/spf13/pflag v1.0.1-0.20171106142849-4c012f6dcd95/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
 github.com/spf13/pflag v1.0.3 h1:zPAT6CGy6wXeQ7NtTnaTerfKOsV6V6F8agHXFiazDkg=
 github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
 github.com/spf13/viper v1.3.2/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DMA2s=
@@ -520,6 +554,7 @@ go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
 go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 golang.org/x/crypto v0.0.0-20170930174604-9419663f5a44/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
+golang.org/x/crypto v0.0.0-20171113213409-9f005a07e0d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
@@ -552,6 +587,8 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20190628185345-da137c7871d7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190827160401-ba9fcec4b297 h1:k7pJ2yAPLPgbskkFdhRCsA77k2fySZ1zf2zCjvQCiIM=
 golang.org/x/net v0.0.0-20190827160401-ba9fcec4b297/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200301022130-244492dfa37a h1:GuSPYbZzB5/dcLNCwLQLsg3obCJtX9IJhpXkvY7kzk0=
+golang.org/x/net v0.0.0-20200301022130-244492dfa37a/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -570,6 +607,7 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190507160741-ecd444e8653b/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190610200419-93c9922d18ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190616124812-15dcb6c0061f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190626221950-04f50cda93cb/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -577,6 +615,8 @@ golang.org/x/sys v0.0.0-20190712062909-fae7ac547cb7/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190813064441-fde4db37ae7a/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200122134326-e047566fdf82 h1:ywK/j/KkyTHcdyYSZNXGjMwgmDSfjglYZ3vStQ/gSCU=
 golang.org/x/sys v0.0.0-20200122134326-e047566fdf82/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527 h1:uYVVQ9WP/Ds2ROhcaGPeIdVq0RIXVLwsHlnvJ+cT1So=
+golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
@@ -599,12 +639,14 @@ google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7
 google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.21.0/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM=
+gopkg.in/airbrake/gobrake.v2 v2.0.9/go.mod h1:/h5ZAUhDkGaJfjzjKLSjv6zCL6O0LLBxU4K+aSYdM/U=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
+gopkg.in/gemnasium/logrus-airbrake-hook.v2 v2.1.2/go.mod h1:Xk6kEKp8OKb+X14hQBKWaSkCsqBpgog8nAV2xsGOxlo=
 gopkg.in/go-playground/assert.v1 v1.2.1/go.mod h1:9RXL0bg/zibRAgZUYszZSwO/z8Y/a8bDuhia5mkpMnE=
 gopkg.in/go-playground/validator.v8 v8.18.2/go.mod h1:RX2a/7Ha8BgOhfk7j780h4/u/RRjR0eouCJSH80/M2Y=
 gopkg.in/go-playground/validator.v9 v9.29.1 h1:SvGtYmN60a5CVKTOzMSyfzWDeZRxRuGvRQyEAKbw1xc=
diff --git a/internal/init.go b/internal/init.go
index 6ac30c0..41b058c 100644
--- a/internal/init.go
+++ b/internal/init.go
@@ -3,10 +3,10 @@ package internal
 import (
 	"flag"
 	"fmt"
-	sentrygin "github.com/getsentry/sentry-go/gin"
 	"github.com/gin-gonic/gin"
-	"github.com/spf13/viper"
 	"github.com/trustwallet/blockatlas/config"
+	"github.com/trustwallet/blockatlas/mq"
+	"github.com/trustwallet/blockatlas/pkg/ginutils"
 	"github.com/trustwallet/blockatlas/pkg/logger"
 	"github.com/trustwallet/blockatlas/storage"
 	"path/filepath"
@@ -19,58 +19,58 @@ var (
 	Date  = time.Now().String()
 )
 
-func InitAPI(defaultPort, defaultConfigPath string) (string, string, *gin.HandlerFunc) {
+func ParseArgs(defaultPort, defaultConfigPath string) (string, string) {
 	var (
 		port, confPath string
-		sg             gin.HandlerFunc
 	)
 
-	LogVersionInfo()
-	sg = sentrygin.New(sentrygin.Options{})
-
 	flag.StringVar(&port, "p", defaultPort, "port for api")
 	flag.StringVar(&confPath, "c", defaultConfigPath, "config file for api")
 	flag.Parse()
 
-	confPath, err := filepath.Abs(confPath)
+	return port, confPath
+}
+
+func InitRedis(host string) *storage.Storage {
+	cache := storage.New()
+	err := cache.Init(host)
 	if err != nil {
 		logger.Fatal(err)
 	}
-
-	config.LoadConfig(confPath)
-	logger.InitLogger()
-
-	return port, confPath, &sg
+	return cache
 }
 
-func InitAPIWithRedis(defaultPort, defaultConfigPath string) (string, string, *gin.HandlerFunc, *storage.Storage) {
-	var (
-		port, confPath string
-		cache          *storage.Storage
-		sg             gin.HandlerFunc
-	)
-	LogVersionInfo()
-	cache = storage.New()
-	sg = sentrygin.New(sentrygin.Options{})
-
-	flag.StringVar(&port, "p", defaultPort, "port for api")
-	flag.StringVar(&confPath, "c", defaultConfigPath, "config file for api")
-	flag.Parse()
-
+func InitConfig(confPath string) {
 	confPath, err := filepath.Abs(confPath)
 	if err != nil {
 		logger.Fatal(err)
 	}
 
 	config.LoadConfig(confPath)
-	logger.InitLogger()
+}
+
+func InitEngine(handler *gin.HandlerFunc, ginMode string) *gin.Engine {
+	gin.SetMode(ginMode)
+	engine := gin.New()
+	engine.Use(ginutils.CheckReverseProxy, *handler)
+	engine.Use(ginutils.CORSMiddleware())
+	engine.Use(gin.Logger())
+
+	engine.OPTIONS("/*path", ginutils.CORSMiddleware())
 
-	err = cache.Init(viper.GetString("storage.redis"))
+	return engine
+}
+
+func InitRabbitMQ(rabbitURI string, prefetchCount int) {
+	err := mq.Init(rabbitURI)
+	if err != nil {
+		logger.Fatal("Failed to init Rabbit MQ", logger.Params{"uri": rabbitURI})
+	}
+	err = mq.Transactions.Declare()
 	if err != nil {
 		logger.Fatal(err)
 	}
-
-	return port, confPath, &sg, cache
+	mq.PrefetchCount = prefetchCount
 }
 
 func LogVersionInfo() {
diff --git a/mq/mq.go b/mq/mq.go
index bcfebb9..c086fdd 100644
--- a/mq/mq.go
+++ b/mq/mq.go
@@ -7,9 +7,10 @@ import (
 )
 
 var (
-	amqpChan *amqp.Channel
-	conn     *amqp.Connection
-	queue    amqp.Queue
+	PrefetchCount int
+	amqpChan      *amqp.Channel
+	conn          *amqp.Connection
+	queue         amqp.Queue
 )
 
 type (
@@ -18,8 +19,10 @@ type (
 )
 
 const (
-	Transactions  Queue = "transactions"
-	Subscriptions Queue = "subscriptions"
+	Transactions         Queue = "transactions"
+	Subscriptions        Queue = "subscriptions"
+	defaultPrefetchCount       = 5
+	minPrefetchCount           = 1
 )
 
 func Init(uri string) (err error) {
@@ -67,11 +70,17 @@ func (q Queue) RunConsumer(consumer Consumer, cache storage.Addresses) {
 		return
 	}
 
+	if PrefetchCount < minPrefetchCount {
+		logger.Info("Change prefetch count to default")
+		PrefetchCount = defaultPrefetchCount
+	}
+
 	err = amqpChan.Qos(
-		20,
+		PrefetchCount,
 		0,
 		true,
 	)
+
 	if err != nil {
 		logger.Error("no qos limit ", err)
 	}


Test patch:
diff --git a/tests/docker_test/data/given_subscriptions_added.json b/tests/docker_test/data/given_subscriptions_added.json
new file mode 100644
index 0000000..6d5efbc
--- /dev/null
+++ b/tests/docker_test/data/given_subscriptions_added.json
@@ -0,0 +1,42 @@
+[
+  {
+    "new_subscriptions": {
+      "60": [
+        "0x0000000000000000000000000000000000000000"
+      ]
+    },
+    "old_subscriptions": null,
+    "guid": "0",
+    "operation": "AddSubscription"
+  },
+  {
+    "new_subscriptions": {
+      "118": [
+        "0x0000000000000000000000000000000000000001"
+      ]
+    },
+    "old_subscriptions": null,
+    "guid": "1",
+    "operation": "AddSubscription"
+  },
+  {
+    "new_subscriptions": {
+      "714": [
+        "0x0000000000000000000000000000000000000002"
+      ]
+    },
+    "old_subscriptions": null,
+    "guid": "2",
+    "operation": "AddSubscription"
+  },
+  {
+    "new_subscriptions": {
+      "144": [
+        "0x0000000000000000000000000000000000000003"
+      ]
+    },
+    "old_subscriptions": null,
+    "guid": "3",
+    "operation": "AddSubscription"
+  }
+]
diff --git a/tests/docker_test/data/wanted_subscriptions_added.json b/tests/docker_test/data/wanted_subscriptions_added.json
new file mode 100644
index 0000000..c89a4b5
--- /dev/null
+++ b/tests/docker_test/data/wanted_subscriptions_added.json
@@ -0,0 +1,22 @@
+[
+  {
+    "coin": 60,
+    "address": "0x0000000000000000000000000000000000000000",
+    "guid": "0"
+  },
+  {
+    "coin": 118,
+    "address": "0x0000000000000000000000000000000000000001",
+    "guid": "1"
+  },
+  {
+    "coin": 714,
+    "address": "0x0000000000000000000000000000000000000002",
+    "guid": "2"
+  },
+  {
+    "coin": 144,
+    "address": "0x0000000000000000000000000000000000000003",
+    "guid": "3"
+  }
+]
diff --git a/tests/docker_test/environment_run_test.go b/tests/docker_test/environment_run_test.go
new file mode 100644
index 0000000..ca13dd7
--- /dev/null
+++ b/tests/docker_test/environment_run_test.go
@@ -0,0 +1,18 @@
+// +build integration
+
+package docker_test
+
+import (
+	"github.com/trustwallet/blockatlas/tests/docker_test/setup"
+	"os"
+	"testing"
+)
+
+func TestMain(m *testing.M) {
+	setup.RunMQContainer()
+	setup.RunRedisContainer()
+	code := m.Run()
+	setup.StopMQContainer()
+	setup.StopRedisContainer()
+	os.Exit(code)
+}
diff --git a/tests/docker_test/observer_test.go b/tests/docker_test/observer_test.go
new file mode 100644
index 0000000..5bccfff
--- /dev/null
+++ b/tests/docker_test/observer_test.go
@@ -0,0 +1,64 @@
+// +build integration
+
+package docker_test
+
+import (
+	"encoding/json"
+	"github.com/stretchr/testify/assert"
+	"github.com/trustwallet/blockatlas/mq"
+	"github.com/trustwallet/blockatlas/pkg/blockatlas"
+	"github.com/trustwallet/blockatlas/services/subscription"
+	"github.com/trustwallet/blockatlas/storage"
+	"github.com/trustwallet/blockatlas/tests/docker_test/setup"
+	"io/ioutil"
+	"path/filepath"
+	"runtime"
+	"testing"
+	"time"
+)
+
+func TestSubscriberAddSubscription(t *testing.T) {
+	_, goFile, _, _ := runtime.Caller(0)
+	testFilePathGiven := filepath.Join(filepath.Dir(goFile), "data", "given_subscriptions_added.json")
+	testFileGiven, err := ioutil.ReadFile(testFilePathGiven)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var givenEvents []blockatlas.SubscriptionEvent
+	if err := json.Unmarshal(testFileGiven, &givenEvents); err != nil {
+		t.Fatal(err)
+	}
+
+	testFilePathWanted := filepath.Join(filepath.Dir(goFile), "data", "wanted_subscriptions_added.json")
+	testFileWanted, err := ioutil.ReadFile(testFilePathWanted)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var wantedEvents []blockatlas.Subscription
+	if err := json.Unmarshal(testFileWanted, &wantedEvents); err != nil {
+		t.Fatal(err)
+	}
+
+	assert.Nil(t, mq.Subscriptions.Declare())
+
+	for _, event := range givenEvents {
+		body, err := json.Marshal(event)
+		assert.Nil(t, err)
+
+		err = mq.Subscriptions.Publish(body)
+		assert.Nil(t, err)
+
+		go mq.Subscriptions.RunConsumer(subscription.Consume, setup.Cache)
+		time.Sleep(time.Second / 5)
+	}
+
+	result, err := setup.Cache.GetAllHM(storage.ATLAS_OBSERVER)
+	assert.Nil(t, err)
+	assert.NotNil(t, result)
+
+	for _, wanted := range wantedEvents {
+		result, err := setup.Cache.Lookup(wanted.Coin, []string{wanted.Address})
+		assert.Nil(t, err)
+		assert.Equal(t, result[0], wanted)
+	}
+}
diff --git a/tests/docker_test/setup/mq.go b/tests/docker_test/setup/mq.go
new file mode 100644
index 0000000..cf7e3ea
--- /dev/null
+++ b/tests/docker_test/setup/mq.go
@@ -0,0 +1,37 @@
+package setup
+
+import (
+	"fmt"
+	"github.com/ory/dockertest"
+	"github.com/trustwallet/blockatlas/mq"
+	"log"
+)
+
+var (
+	mqResource     *dockertest.Resource
+)
+
+func runMQContainer() error {
+	var err error
+	pool, err := dockertest.NewPool("")
+	if err != nil {
+		log.Fatalf("Could not connect to docker: %s", err)
+	}
+
+	mqResource, err = pool.Run("rabbitmq", "latest", nil)
+	if err != nil {
+		log.Fatalf("Could not start resource: %s", err)
+	}
+
+	if err = pool.Retry(func() error {
+		return mq.Init(fmt.Sprintf("amqp://localhost:%s", mqResource.GetPort("5672/tcp")))
+	}); err != nil {
+		return err
+	}
+	return nil
+}
+
+func stopMQContainer() error {
+	mq.Close()
+	return mqResource.Close()
+}
\ No newline at end of file
diff --git a/tests/docker_test/setup/redis.go b/tests/docker_test/setup/redis.go
new file mode 100644
index 0000000..05b6929
--- /dev/null
+++ b/tests/docker_test/setup/redis.go
@@ -0,0 +1,43 @@
+package setup
+
+import (
+	"fmt"
+	"github.com/ory/dockertest"
+	"github.com/trustwallet/blockatlas/storage"
+	"log"
+)
+
+var (
+	Cache *storage.Storage
+	redisResource     *dockertest.Resource
+)
+
+func runRedisContainer() error {
+	var err error
+	pool, err := dockertest.NewPool("")
+	if err != nil {
+		log.Fatalf("Could not connect to docker: %s", err)
+	}
+
+	redisResource, err = pool.Run("redis", "latest", nil)
+	if err != nil {
+		log.Fatalf("Could not start resource: %s", err)
+	}
+
+	if err = pool.Retry(func() error {
+		var err error
+		Cache = storage.New()
+		err = Cache.Init(fmt.Sprintf("redis://localhost:%s", redisResource.GetPort("6379/tcp")))
+		if err != nil {
+			return err
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+	return nil
+}
+
+func stopRedisContainer() error {
+	return redisResource.Close()
+}
\ No newline at end of file
diff --git a/tests/docker_test/setup/setup.go b/tests/docker_test/setup/setup.go
new file mode 100644
index 0000000..80daf05
--- /dev/null
+++ b/tests/docker_test/setup/setup.go
@@ -0,0 +1,27 @@
+package setup
+
+import "log"
+
+func RunRedisContainer(){
+	if err := runRedisContainer(); err != nil{
+		log.Fatal(err)
+	}
+}
+
+func StopRedisContainer(){
+	if err := stopRedisContainer(); err != nil{
+		log.Fatal(err)
+	}
+}
+
+func RunMQContainer(){
+	if err := runMQContainer(); err != nil{
+		log.Fatal(err)
+	}
+}
+
+func StopMQContainer(){
+	if err := stopMQContainer(); err != nil{
+		log.Fatal(err)
+	}
+}
\ No newline at end of file
diff --git a/tests/integration/domains/domains.go b/tests/integration/domains/domains.go
index a5dac9e..058e048 100644
--- a/tests/integration/domains/domains.go
+++ b/tests/integration/domains/domains.go
@@ -79,15 +79,15 @@ func TestDomains(t *testing.T) {
 			},
 			false,
 		},
-		{
-			"test @fiotestnet domain",
-			"adam@fiotestnet",
-			[]uint64{coin.ETH},
-			[]blockatlas.Resolved{
-				{Result: "0xce5cB6c92Da37bbBa91Bd40D4C9D4D724A3a8F51", Coin: coin.ETH},
-			},
-			false,
-		},
+		//{
+		//	"test @fiotestnet domain",
+		//	"adam@fiotestnet",
+		//	[]uint64{coin.ETH},
+		//	[]blockatlas.Resolved{
+		//		{Result: "0xce5cB6c92Da37bbBa91Bd40D4C9D4D724A3a8F51", Coin: coin.ETH},
+		//	},
+		//	false,
+		//},
 		{
 			"test batch .crypto domains",
 			"dpantani.crypto",