Initial commit

storage/mongodb.go

@@ -0,0 +1,122 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/spf13/viper"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+	"gopkg.in/mgo.v2/bson"
+)
+
+type MongoDBStorage struct {
+	*mongo.Client
+	DbName      string
+	Collections map[string]string
+}
+
+func NewMongoDBStorage(cfg *viper.Viper) (MongoDBStorage, error) {
+	var (
+		mongodb_host   = cfg.GetString("storage.db.mongodb.host")
+		mongodb_port   = cfg.GetString("storage.db.mongodb.port")
+		mongodb_dbname = cfg.GetString("storage.db.mongodb.db_name")
+		mongodb_events = cfg.GetString("storage.db.mongodb.collections.events")
+	)
+
+	client, err := mongo.NewClient(options.Client().ApplyURI("mongodb://" + mongodb_host + ":" + mongodb_port))
+	if err != nil {
+		return MongoDBStorage{}, err
+	}
+
+	err = client.Connect(context.TODO())
+
+	if err != nil {
+		return MongoDBStorage{}, err
+	}
+
+	storage := MongoDBStorage{
+		Client: client,
+		DbName: mongodb_dbname,
+		Collections: map[string]string{
+			"events": mongodb_events,
+		},
+	}
+	//TODO Indexes
+	return storage, err
+}
+
+func (s MongoDBStorage) CreateEvent(event Event) error {
+	collection := s.Client.Database(s.DbName).Collection(s.Collections["events"])
+	if _, err := collection.InsertOne(context.TODO(), event); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (s MongoDBStorage) GetEvent(id string) (*Event, error) {
+	collection := s.Client.Database(s.DbName).Collection(s.Collections["events"])
+
+	event := &Event{}
+	if err := collection.FindOne(context.TODO(), bson.M{"_id": id}).Decode(event); err != nil {
+		return nil, err
+	}
+
+	return event, nil
+}
+
+func (s MongoDBStorage) GetEvents(namespaces []string) (events []Event, err error) {
+	collection := s.Client.Database(s.DbName).Collection(s.Collections["events"])
+
+	var cur *mongo.Cursor
+
+	findOptions := options.Find()
+	//findOptions.SetSort(bson.D{{"startdate", 1}})
+
+	if len(namespaces) == 0 {
+		cur, err = collection.Find(context.TODO(), bson.D{}, findOptions)
+		if err != nil {
+			fmt.Println("...")
+			fmt.Println(err)
+			return events, err
+		}
+	} else {
+		cur, err = collection.Find(context.TODO(), bson.M{"namespace": bson.M{"$in": namespaces}}, findOptions)
+		if err != nil {
+			fmt.Println("Error here")
+			fmt.Println(err)
+			return events, err
+		}
+	}
+
+	for cur.Next(context.TODO()) {
+		var event Event
+		//var elem bson.M
+
+		if err := cur.Decode(&event); err != nil {
+			return events, err
+		}
+
+		// fmt.Println(elem)
+
+		// bsonBytes, _ := bson.Marshal(elem)
+		// fmt.Println(string(bsonBytes))
+		// bson.Unmarshal(bsonBytes, &event)
+
+		events = append(events, event)
+
+	}
+
+	return
+}
+
+func (s MongoDBStorage) AddSubscriber(eventid string, subscriber string) error {
+	collection := s.Client.Database(s.DbName).Collection(s.Collections["events"])
+
+	filter := bson.M{"_id": eventid}
+	push := bson.M{"$push": bson.M{"subscribers": subscriber}}
+
+	_, err := collection.UpdateOne(context.TODO(), filter, push)
+	return err
+}