mongo

Author: Enapiuz

keeper/keeper.go

@@ -0,0 +1,32 @@
+package keeper
+
+import (
+"gopkg.in/mgo.v2"
+)
+
+const SubscribersDatabase string = "subscribers"
+
+type Keeper struct {
+mongo *mgo.Session
+databaseName string
+}
+
+func NewKeeper(session *mgo.Session) *Keeper {
+return &Keeper{session, SubscribersDatabase}
+}
+
+func (k *Keeper) SetCustomDatabaseName(name string) {
+k.databaseName = name
+}
+
+func (k *Keeper) StoreSubscriber(pluginName string, data interface{}) error {
+return k.mongo.DB(k.databaseName).C("stored_" + pluginName).Insert(data)
+}
+
+func (k *Keeper) RemoveSubscriber(pluginName string, data interface{}) error {
+return k.mongo.DB(k.databaseName).C("stored_" + pluginName).Remove(data)
+}
+
+func (k *Keeper) GetSubscribers(pluginName string, data interface{}) error {
+return k.mongo.DB(k.databaseName).C("stored_" + pluginName).Find(nil).All(data)
+}

keeper/master.go

@@ -39,10 +39,10 @@ func main() {
 panic(err)
 }
 
-master := keeper.NewMongoMaster(msession)
+kpr := keeper.NewKeeper(msession)
 
 pluginStorage := plugins.NewPluginStorage()
-pluginStorage.Register(web.NewPluginWeb(master.GetCollection("web")))
+pluginStorage.Register(web.NewPluginWeb(kpr))
 pluginStorage.Register(logPlugin.NewLog())
 
 eventService := event.NewEventService(eventRmq)

main.go

@@ -10,33 +10,32 @@ import (
 
 log "github.com/Sirupsen/logrus"
 "github.com/qa-dev/universe/event"
+"github.com/qa-dev/universe/keeper"
 "github.com/qa-dev/universe/plugins"
-"gopkg.in/mgo.v2"
-"gopkg.in/mgo.v2/bson"
 )
 
-const PLUGIN_TAG string = "web"
+const PluginTag string = "web"
 
 type PluginWeb struct {
-client  HttpRequester
-collection *mgo.Collection
+client HttpRequester
+keeper *keeper.Keeper
 }
 
 type HttpRequester interface {
 Do(req *http.Request) (*http.Response, error)
 }
 
-func NewPluginWeb(collection *mgo.Collection) *PluginWeb {
+func NewPluginWeb(kpr *keeper.Keeper) *PluginWeb {
 httpClient := &http.Client{
 Timeout: time.Second * 10,
 }
-return &PluginWeb{httpClient, collection}
+return &PluginWeb{httpClient, kpr}
 }
 
 func (p *PluginWeb) GetPluginInfo() *plugins.PluginInfo {
 return &plugins.PluginInfo{
 Name: "Web",
-Tag: PLUGIN_TAG,
+Tag: PluginTag,
 Version: 1,
 }
 }
@@ -49,7 +48,7 @@ func (p *PluginWeb) Subscribe(input []byte) error {
 errorText := fmt.Sprintf("Invalid input data: %s", string(input))
 return errors.New(errorText)
 }
-return p.collection.Insert(subscribeData)
+return p.keeper.StoreSubscriber(PluginTag, subscribeData)
 }
 
 func (p *PluginWeb) Unsubscribe(input []byte) error {
@@ -60,12 +59,12 @@ func (p *PluginWeb) Unsubscribe(input []byte) error {
 errorText := fmt.Sprintf("Invalid input data: %s", string(input))
 return errors.New(errorText)
 }
-return p.collection.Remove(unsubscribeData)
+return p.keeper.RemoveSubscriber(PluginTag, unsubscribeData)
 }
 
 func (p *PluginWeb) ProcessEvent(eventData event.Event) {
 var result []SubscribeData
-p.collection.Find(bson.M{"eventname": eventData.Name}).All(&result)
+p.keeper.GetSubscribers(PluginTag, &result)
 for _, data := range result {
 go p.sendRequest(data.Url, eventData.Payload)
 }
@@ -82,16 +81,4 @@ func (p *PluginWeb) sendRequest(url string, payload []byte) {
 p.client.Do(req)
 }
 
-func (p *PluginWeb) Loaded() {
-index := mgo.Index{
-Key: []string{"eventname", "url"},
-Unique: true,
-DropDups: true,
-Background: true,
-Sparse: true,
-}
-err := p.collection.EnsureIndex(index)
-if err != nil {
-panic(err)
-}
-}
+func (p *PluginWeb) Loaded() {}

plugins/web/plugin.go

@@ -10,9 +10,9 @@ import (
 "time"
 
 "github.com/qa-dev/universe/event"
+"github.com/qa-dev/universe/keeper"
 "github.com/stretchr/testify/assert"
 "gopkg.in/mgo.v2"
-"gopkg.in/mgo.v2/bson"
 )
 
 type FakeClosingBuffer struct {
@@ -44,6 +44,7 @@ func (c FakePostClient) Do(r *http.Request) (*http.Response, error) {
 return response, nil
 }
 
+var kpr *keeper.Keeper
 var database *mgo.Database
 
 func init() {
@@ -53,85 +54,87 @@ func init() {
 panic("cant connect mongo")
 }
 database = session.DB("test_plugin_web")
-database.DropDatabase()
+kpr = keeper.NewKeeper(session)
+kpr.SetCustomDatabaseName("test_plugin_web")
 }
 
 func TestNewPluginWeb(t *testing.T) {
-c := database.C("test_new_plugin_web")
-p := NewPluginWeb(c)
-assert.NotNil(t, p.collection)
+p := NewPluginWeb(kpr)
+assert.NotNil(t, p.keeper)
 }
 
 func TestPluginWeb_GetPluginInfo(t *testing.T) {
-c := database.C("test_plugin_web_get_plugin_info")
-p := NewPluginWeb(c)
+p := NewPluginWeb(kpr)
 assert.Equal(t, "web", p.GetPluginInfo().Tag)
 assert.Equal(t, "Web", p.GetPluginInfo().Name)
 }
 
 func TestPluginWeb_Subscribe(t *testing.T) {
-c := database.C("test_plugin_web_subscribe")
-p := NewPluginWeb(c)
+database.DropDatabase()
+p := NewPluginWeb(kpr)
 inJson := []byte(`{"event_name": "test", "url": "hello"}`)
 err := p.Subscribe(inJson)
 assert.NoError(t, err)
-resCount, err := c.Find(bson.M{"eventname": "test"}).Count()
+var result []SubscribeData
+err = kpr.GetSubscribers(PluginTag, &result)
 assert.NoError(t, err)
-assert.Equal(t, 1, resCount)
+assert.Equal(t, 1, len(result))
 }
 
 func TestPluginWeb_Subscribe_WrongInput(t *testing.T) {
-c := database.C("test_plugin_web_wrong_input")
-p := NewPluginWeb(c)
+p := NewPluginWeb(kpr)
 inJson := []byte("{}")
 err := p.Subscribe(inJson)
 assert.Error(t, err)
 }
 
 func TestPluginWeb_Unsubscribe(t *testing.T) {
-c := database.C("test_plugin_web_unsubscribe")
-p := NewPluginWeb(c)
+database.DropDatabase()
+p := NewPluginWeb(kpr)
 inJson := []byte(`{"event_name": "test", "url": "hello"}`)
 err := p.Subscribe(inJson)
 assert.NoError(t, err)
-resCount, err := c.Find(bson.M{"eventname": "test"}).Count()
+var result []SubscribeData
+err = kpr.GetSubscribers(PluginTag, &result)
 assert.NoError(t, err)
-assert.Equal(t, 1, resCount)
+assert.Equal(t, 1, len(result))
 err = p.Unsubscribe(inJson)
 assert.NoError(t, err)
-resCount, err = c.Find(bson.M{"eventname": "test"}).Count()
+result = nil
+err = kpr.GetSubscribers(PluginTag, &result)
 assert.NoError(t, err)
-assert.Equal(t, 0, resCount)
+assert.Equal(t, 0, len(result))
 }
 
 func TestPluginWeb_Unsubscribe_WrongInput(t *testing.T) {
-c := database.C("test_plugin_web_unsubscribe_wrong_input")
-p := NewPluginWeb(c)
+p := NewPluginWeb(kpr)
 inJson := []byte("{}")
 err := p.Unsubscribe(inJson)
 assert.Error(t, err)
 }
 
 func TestPluginWeb_Unsubscribe_NonExistentSubscriber(t *testing.T) {
-c := database.C("test_plugin_web_non_existent_subscriber")
-p := NewPluginWeb(c)
+database.DropDatabase()
+p := NewPluginWeb(kpr)
 subscribeJson := []byte(`{"event_name": "test", "url": "hello"}`)
 err := p.Subscribe(subscribeJson)
 assert.NoError(t, err)
-resCount, err := c.Find(bson.M{"eventname": "test"}).Count()
+var result []SubscribeData
+err = kpr.GetSubscribers(PluginTag, &result)
 assert.NoError(t, err)
-assert.Equal(t, 1, resCount)
+assert.Equal(t, 1, len(result))
 unsubscribeJson := []byte(`{"event_name": "test", "url": "bye"}`)
 err = p.Unsubscribe(unsubscribeJson)
 assert.Error(t, err)
-resCount, err = c.Find(bson.M{"eventname": "test"}).Count()
+result = nil
+err = kpr.GetSubscribers(PluginTag, &result)
 assert.NoError(t, err)
-assert.Equal(t, 1, resCount)
+assert.Equal(t, 1, len(result))
 }
 
 func TestPluginWeb_ProcessEvent(t *testing.T) {
-c := database.C("test_plugin_web_process_event")
-p := NewPluginWeb(c)
+database.DropDatabase()
+p := NewPluginWeb(kpr)
 expectedUrl := "test_url"
 expectedData := []byte(`{"hello": "world"}`)
 fakeClient := FakePostClient{t, expectedUrl, expectedData}