Merge branch 'reinvent-rabbit' into mongo+rabbit

# Conflicts: #	.travis.yml #	main.go #	plugins/plugin.go #	plugins/storage_test.go #	plugins/web/plugin.go

Author: Enapiuz

dispatcher/dispatcher.go

@@ -4,45 +4,56 @@ import (
 "encoding/json"
 "net/http"
 
+log "github.com/Sirupsen/logrus"
 "github.com/qa-dev/universe/event"
 "github.com/qa-dev/universe/plugins"
-"github.com/qa-dev/universe/rabbitmq"
+"github.com/qa-dev/universe/queue"
 )
 
 type ClientInterface interface {
 Do(req *http.Request) (*http.Response, error)
 }
 
 type Dispatcher struct {
-rmq  *rabbitmq.RabbitMQ
+queue *queue.Queue
 pluginStorage *plugins.PluginStorage
 }
 
-func NewDispatcher(rmq *rabbitmq.RabbitMQ, storage *plugins.PluginStorage) *Dispatcher {
-return &Dispatcher{rmq, storage}
+func NewDispatcher(queue *queue.Queue, storage *plugins.PluginStorage) *Dispatcher {
+return &Dispatcher{queue, storage}
 }
 
 func (d *Dispatcher) Run() {
 go d.worker()
 }
 
 func (d *Dispatcher) worker() {
-consumeObj, err := d.rmq.Consume("consumer")
+msgs, err := d.queue.GetConsumer("consumer")
 if err != nil {
-panic(err)
+log.Error("Error get consumer in event dispatcher worker")
 }
-
 for {
-rawData := <-consumeObj
-var e event.Event
-err = json.Unmarshal(rawData.Body(), &e)
+if d.queue.IsOnline() == false {
+log.Info("Worker lost connection to queue. Waiting...")
+backOnlineChan := make(chan *error)
+d.queue.NotifyReconnect(backOnlineChan)
+_ = <-backOnlineChan
+newMsgs, err := d.queue.GetConsumer("consumer")
+if err != nil {
+log.Error("Error get consumer in event dispatcher worker")
+}
+msgs = newMsgs
+log.Info("Worker established connection to queue")
+}
+data := <-msgs
+var ev event.Event
+err = json.Unmarshal(data.Body(), &ev)
 if err != nil {
-// TODO log?
-panic(err)
+data.Reject()
+log.Error("Error unmarchal event in event dispatcher worker")
+continue
 }
-
-d.pluginStorage.ProcessEvent(e)
-
-rawData.Ack(false)
+d.pluginStorage.ProcessEvent(&ev)
+data.Ack()
 }
 }

dispatcher/dispatcher_test.go

@@ -8,11 +8,10 @@ import (
 "net/http"
 "os"
 "testing"
-"time"
 
-log "github.com/Sirupsen/logrus"
 "github.com/qa-dev/universe/event"
 "github.com/qa-dev/universe/plugins"
+"github.com/qa-dev/universe/queue"
 "github.com/qa-dev/universe/rabbitmq"
 "github.com/qa-dev/universe/subscribe"
 "github.com/stretchr/testify/assert"
@@ -23,7 +22,7 @@ var amqpUri string
 func init() {
 amqpUri = os.Getenv("AMQP_URI")
 if amqpUri == "" {
-log.Fatal("AMQP_URI is required to run rabbitmq tests")
+amqpUri = "amqp://guest:guest@127.0.0.1:5672/"
 }
 }
 
@@ -57,30 +56,26 @@ func (c *FakePostClient) Do(r *http.Request) (*http.Response, error) {
 }
 
 func TestNewDispatcher(t *testing.T) {
-rmq := rabbitmq.NewRabbitMQ(amqpUri, "test_event_service_push_event_queue")
-defer rmq.Close()
+rmq := rabbitmq.NewRabbitMQ(amqpUri, "test_new_dispatcher")
+q := queue.NewQueue(rmq)
 storage := plugins.NewPluginStorage()
-// Даем время на подключение
-time.Sleep(5 * time.Second)
-dsp := NewDispatcher(rmq, storage)
-assert.Equal(t, fmt.Sprintf("%p", rmq), fmt.Sprintf("%p", dsp.rmq))
+dsp := NewDispatcher(q, storage)
+assert.Equal(t, fmt.Sprintf("%p", q), fmt.Sprintf("%p", dsp.queue))
 }
 
 func TestDispatcher_Run(t *testing.T) {
-rmq := rabbitmq.NewRabbitMQ(amqpUri, "test_event_service_push_event_queue")
-defer rmq.Close()
+rmq := rabbitmq.NewRabbitMQ(amqpUri, "test_new_dispatcher")
+q := queue.NewQueue(rmq)
 storage := plugins.NewPluginStorage()
-// Даем время на подключение
-time.Sleep(5 * time.Second)
 requestData := []byte(`{"test": "test"}`)
 subscrService := subscribe.NewSubscribeService(storage)
-eventService := event.NewEventService(rmq)
+eventService := event.NewEventService(q)
 subscribeData := []byte(`{"test": "hello"}`)
 subscrService.ProcessSubscribe("log", subscribeData)
-dsp := NewDispatcher(rmq, storage)
+dsp := NewDispatcher(q, storage)
 assert.NotNil(t, dsp)
 dsp.Run()
-err := eventService.Publish(event.Event{"test.event", requestData})
+err := eventService.Publish(&event.Event{"test.event", requestData})
 assert.NoError(t, err)
 // TODO: assert log
 }

event/service.go

@@ -4,23 +4,23 @@ import (
 "errors"
 
 log "github.com/Sirupsen/logrus"
-"github.com/qa-dev/universe/rabbitmq"
+"github.com/qa-dev/universe/queue"
 )
 
 type EventService struct {
-rmq *rabbitmq.RabbitMQ
+queue *queue.Queue
 }
 
-func NewEventService(rmq *rabbitmq.RabbitMQ) *EventService {
-return &EventService{rmq}
+func NewEventService(queue *queue.Queue) *EventService {
+return &EventService{queue}
 }
 
-func (e *EventService) Publish(ev Event) error {
+func (e *EventService) Publish(ev *Event) error {
 if ev.Name == "" {
 log.Println("Got blank event name")
 return errors.New("BLANK EVENT NAME")
 }
 log.Println("Got event name", ev.Name)
-e.rmq.PublishWithPriority(ev, 1)
+e.queue.SendEvent(ev)
 return nil
 }

event/service_test.go

@@ -7,7 +7,7 @@ import (
 "testing"
 "time"
 
-log "github.com/Sirupsen/logrus"
+"github.com/qa-dev/universe/queue"
 "github.com/qa-dev/universe/rabbitmq"
 "github.com/stretchr/testify/assert"
 )
@@ -17,47 +17,47 @@ var amqpUri string
 func init() {
 amqpUri = os.Getenv("AMQP_URI")
 if amqpUri == "" {
-log.Fatal("AMQP_URI is required to run rabbitmq tests")
+amqpUri = "amqp://guest:guest@127.0.0.1:5672/"
 }
 }
 
 func TestNewEventService(t *testing.T) {
-rmq := &rabbitmq.RabbitMQ{}
-es := NewEventService(rmq)
-assert.Equal(t, fmt.Sprintf("%p", rmq), fmt.Sprintf("%p", es.rmq))
+rmq := rabbitmq.NewRabbitMQ(amqpUri, "test_new_dispatcher")
+time.Sleep(2 * time.Second)
+q := queue.NewQueue(rmq)
+es := NewEventService(q)
+assert.Equal(t, fmt.Sprintf("%p", q), fmt.Sprintf("%p", es.queue))
 }
 
 func TestEventService_PushEvent(t *testing.T) {
-rmq := rabbitmq.NewRabbitMQ(amqpUri, "test_event_service_push_event_queue")
-defer rmq.Close()
-// Даем время на подключение
-time.Sleep(5 * time.Second)
-es := NewEventService(rmq)
-
-consumeObj, err := rmq.Consume("test_consumer")
-assert.NoError(t, err)
+rmq := rabbitmq.NewRabbitMQ(amqpUri, "test_new_dispatcher")
+time.Sleep(2 * time.Second)
+q := queue.NewQueue(rmq)
+es := NewEventService(q)
 
 go func() {
-raw := <-consumeObj
+msgs, err := q.GetConsumer("test_consumer")
+assert.NoError(t, err)
+data := <-msgs
 var e Event
-err = json.Unmarshal(raw.Body(), &e)
+err = json.Unmarshal(data.Body(), &e)
+data.Ack()
 assert.NoError(t, err)
 assert.Equal(t, "test.event", e.Name, "Wrong event name generated")
 }()
 
-err = es.Publish(Event{"test.event", []byte("test")})
+err := es.Publish(&Event{"test.event", []byte("test")})
 assert.NoError(t, err)
 time.Sleep(1 * time.Second)
 }
 
 func TestEventService_PushEvent_Blank(t *testing.T) {
-rmq := rabbitmq.NewRabbitMQ(amqpUri, "test_event_service_push_event_queue")
-defer rmq.Close()
-// Даем время на подключение
-time.Sleep(5 * time.Second)
-es := NewEventService(rmq)
+rmq := rabbitmq.NewRabbitMQ(amqpUri, "test_new_dispatcher")
+time.Sleep(2 * time.Second)
+q := queue.NewQueue(rmq)
+es := NewEventService(q)
 
-err := es.Publish(Event{"", []byte("test")})
+err := es.Publish(&Event{"", []byte("test")})
 
 assert.Error(t, err)
 assert.Equal(t, "BLANK EVENT NAME", err.Error())

handlers/event.go

@@ -14,7 +14,7 @@ type EventHandler struct {
 }
 
 type EventPublisher interface {
-Publish(event.Event) error
+Publish(*event.Event) error
 }
 
 func NewEventHandler(eventService EventPublisher) *EventHandler {
@@ -36,7 +36,7 @@ func (h *EventHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 }
 
 e := event.Event{eventName, payload}
-err = h.eventService.Publish(e)
+err = h.eventService.Publish(&e)
 if err != nil {
 resp.WriteHeader(http.StatusInternalServerError)
 resp.Write([]byte(`{"error": ` + strconv.Quote(err.Error()) + `}`))

main.go

@@ -4,7 +4,6 @@ import (
 "flag"
 "fmt"
 "net/http"
-"time"
 
 log "github.com/Sirupsen/logrus"
 "github.com/qa-dev/universe/config"
@@ -15,9 +14,10 @@ import (
 "github.com/qa-dev/universe/plugins"
 logPlugin "github.com/qa-dev/universe/plugins/log"
 "github.com/qa-dev/universe/plugins/web"
-"github.com/qa-dev/universe/rabbitmq"
 "github.com/qa-dev/universe/subscribe"
-"gopkg.in/mgo.v2"
+"github.com/qa-dev/universe/rabbitmq"
+"time"
+"github.com/qa-dev/universe/queue"
 )
 
 func main() {
@@ -34,6 +34,8 @@ func main() {
 time.Sleep(2 * time.Second)
 defer eventRmq.Close()
 
+eventQueue := queue.NewQueue(eventRmq)
+
 msession, err := mgo.Dial(cfg.Mongo.Host + ":" + cfg.Mongo.Port)
 if err != nil {
 panic(err)
@@ -45,9 +47,9 @@ func main() {
 pluginStorage.Register(web.NewPluginWeb(kpr))
 pluginStorage.Register(logPlugin.NewLog())
 
-eventService := event.NewEventService(eventRmq)
+eventService := event.NewEventService(eventQueue)
 subscribeService := subscribe.NewSubscribeService(pluginStorage)
-dispatcherService := dispatcher.NewDispatcher(eventRmq, pluginStorage)
+dispatcherService := dispatcher.NewDispatcher(eventQueue, pluginStorage)
 dispatcherService.Run()
 
 for _, plg := range pluginStorage.GetPlugins() {

plugins/log/log.go

@@ -22,7 +22,7 @@ func (l Log) GetPluginInfo() *plugins.PluginInfo {
 }
 }
 
-func (l Log) ProcessEvent(eventData event.Event) {
+func (l Log) ProcessEvent(eventData *event.Event) {
 l.logger.Info(eventData)
 }
 

plugins/log/log_test.go

@@ -18,7 +18,7 @@ func TestLog_ProcessEvent(t *testing.T) {
 l.logger.Out = &b
 o.Register(l)
 
-o.ProcessEvent(event.Event{"Event", []byte(`{"hello": "test"}`)})
+o.ProcessEvent(&event.Event{"Event", []byte(`{"hello": "test"}`)})
 time.Sleep(200 * time.Millisecond)
 
 t.Log(b.String())

plugins/plugin.go

@@ -8,7 +8,7 @@ type Plugin interface {
 GetPluginInfo() *PluginInfo
 Subscribe(input []byte) error
 Unsubscribe(input []byte) error
-ProcessEvent(eventData event.Event)
+ProcessEvent(eventData *event.Event)
 Loaded()
 }
 

plugins/storage.go

@@ -19,11 +19,9 @@ func (o *PluginStorage) Register(v Plugin) {
 o.plugins = append(o.plugins, v)
 }
 
-func (o *PluginStorage) ProcessEvent(eventData event.Event) {
+func (o *PluginStorage) ProcessEvent(eventData *event.Event) {
 for _, ob := range o.plugins {
-go func(o Plugin) {
-o.ProcessEvent(eventData)
-}(ob)
+ob.ProcessEvent(eventData)
 }
 }