1
0
mirror of https://github.com/mainflux/mainflux.git synced 2025-04-29 13:49:28 +08:00
Darko Draskovic 2b393ad50f MF-237 - Add support for storing messages in MongoDB (#307)
* Add mongodb-writer

Signed-off-by: Darko Draskovic <darko.draskovic@gmail.com>

* Add official mongodb driver

Signed-off-by: Darko Draskovic <darko.draskovic@gmail.com>

* Move Connect to main.go

Signed-off-by: Darko Draskovic <darko.draskovic@gmail.com>

* Remove bson.NewDoc and write msg directly in db

Signed-off-by: Darko Draskovic <darko.draskovic@gmail.com>

* Add MongoDB writer tests

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Update README.md

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Add mongodb services compose to addons dir

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Update docs

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Update docs and tests

Refactor code.

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Expose MetricsMiddleware to align writers with other services

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Add logging middleware

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

* Update load tests version

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>
2018-06-01 15:50:23 +02:00

130 lines
3.6 KiB
Go

// Copyright (C) MongoDB, Inc. 2017-present.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License. You may obtain
// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
package mongo
import (
"bytes"
"fmt"
"github.com/mongodb/mongo-go-driver/bson"
"github.com/mongodb/mongo-go-driver/core/result"
)
// InsertOneResult is a result of an InsertOne operation.
//
// InsertedID will be a Go type that corresponds to a BSON type.
type InsertOneResult struct {
// The identifier that was inserted.
InsertedID interface{}
}
// InsertManyResult is a result of an InsertMany operation.
type InsertManyResult struct {
// Maps the indexes of inserted documents to their _id fields.
InsertedIDs []interface{}
}
// DeleteResult is a result of an DeleteOne operation.
type DeleteResult struct {
// The number of documents that were deleted.
DeletedCount int64 `bson:"n"`
}
// ListDatabasesResult is a result of a ListDatabases operation. Each specification
// is a description of the datbases on the server.
type ListDatabasesResult struct {
Databases []DatabaseSpecification
TotalSize int64
}
func (ldr ListDatabasesResult) fromResult(res result.ListDatabases) ListDatabasesResult {
ldr.Databases = make([]DatabaseSpecification, 0, len(res.Databases))
for _, spec := range res.Databases {
ldr.Databases = append(
ldr.Databases,
DatabaseSpecification{Name: spec.Name, SizeOnDisk: spec.SizeOnDisk, Empty: spec.Empty},
)
}
ldr.TotalSize = res.TotalSize
return ldr
}
// DatabaseSpecification is the information for a single database returned
// from a ListDatabases operation.
type DatabaseSpecification struct {
Name string
SizeOnDisk int64
Empty bool
}
// UpdateResult is a result of an update operation.
//
// UpsertedID will be a Go type that corresponds to a BSON type.
type UpdateResult struct {
// The number of documents that matched the filter.
MatchedCount int64
// The number of documents that were modified.
ModifiedCount int64
// The identifier of the inserted document if an upsert took place.
UpsertedID interface{}
}
// UnmarshalBSON implements the bson.Unmarshaler interface.
func (result *UpdateResult) UnmarshalBSON(b []byte) error {
itr, err := bson.Reader(b).Iterator()
if err != nil {
return err
}
for itr.Next() {
elem := itr.Element()
switch elem.Key() {
case "n":
switch elem.Value().Type() {
case bson.TypeInt32:
result.MatchedCount = int64(elem.Value().Int32())
case bson.TypeInt64:
result.MatchedCount = elem.Value().Int64()
default:
return fmt.Errorf("Received invalid type for n, should be Int32 or Int64, received %s", elem.Value().Type())
}
case "nModified":
switch elem.Value().Type() {
case bson.TypeInt32:
result.ModifiedCount = int64(elem.Value().Int32())
case bson.TypeInt64:
result.ModifiedCount = elem.Value().Int64()
default:
return fmt.Errorf("Received invalid type for nModified, should be Int32 or Int64, received %s", elem.Value().Type())
}
case "upserted":
switch elem.Value().Type() {
case bson.TypeArray:
e, err := elem.Value().ReaderArray().ElementAt(0)
if err != nil {
break
}
if e.Value().Type() != bson.TypeEmbeddedDocument {
break
}
var d struct {
ID interface{} `bson:"_id"`
}
err = bson.NewDecoder(bytes.NewReader(e.Value().ReaderDocument())).Decode(&d)
if err != nil {
return err
}
result.UpsertedID = d.ID
default:
return fmt.Errorf("Received invalid type for upserted, should be Array, received %s", elem.Value().Type())
}
}
}
return nil
}