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) AddSubscription(eventid string, subscription Subscription) error { collection := s.Client.Database(s.DbName).Collection(s.Collections["events"]) filter := bson.M{"_id": eventid} push := bson.M{"$push": bson.M{"subscriptions": subscription}} _, err := collection.UpdateOne(context.TODO(), filter, push) return err } func (s MongoDBStorage) UpdateSubscription(eventid string, subscriber string, deletesubscription Subscription) error { collection := s.Client.Database(s.DbName).Collection(s.Collections["events"]) event := &Event{} if errr := collection.FindOne(context.TODO(), bson.M{"_id": eventid}).Decode(event); errr != nil { return errr } for i := range event.Subscriptions { if event.Subscriptions[i].Subscriber == subscriber { filter := bson.M{"_id": eventid} push := bson.M{"$push": bson.M{"deletedsubscriptions": deletesubscription}} pull := bson.M{"$pull": bson.M{"subscriptions": bson.M{"subscriber": subscriber}}} _, er := collection.UpdateOne(context.TODO(), filter, push) if _, err := collection.UpdateOne(context.TODO(), filter, pull); err != nil { return err } return er } } func (s MongoDBStorage) UpdateEvent(event Event) error { collection := s.Client.Database(s.DbName).Collection(s.Collections["events"]) if _, err := collection.ReplaceOne(context.TODO(), bson.M{"_id": event.ID}, event); err != nil { fmt.Println(err) return err } return nil }