feature/ext-mongo (#177)

* externalize mongo wrapper * remove mongo env from drone * export mongo test for coverage report
2018-07-22 22:46:25 -04:00
parent fdf161850b
commit dbd1d4069f
27 changed files with 311 additions and 432 deletions
@@ -0,0 +1,12 @@
+# Binaries for programs and plugins
+*.exe
+*.exe~
+*.dll
+*.so
+*.dylib
+
+# Test binary, build with `go test -c`
+*.test
+
+# Output of the go coverage tool, specifically when used with LiteIDE
+*.out
@@ -0,0 +1,23 @@
+language: go
+
+go:
+  - "1.10.x"
+
+go_import_path: github.com/go-pkgz/mongo
+
+services: mongodb
+
+before_install:
+  - cp -rf .vendor vendor
+  - go get github.com/mattn/goveralls
+  - go get gopkg.in/alecthomas/gometalinter.v2
+  - $GOPATH/bin/gometalinter.v2 --install
+
+before_script:
+  - sleep 3
+
+script:
+  - export MONGO_TEST=mongodb://127.0.0.1:27017
+  - go test ./...
+  - $GOPATH/bin/gometalinter.v2 --exclude=test --exclude=mock --exclude=vendor ./...
+  - $GOPATH/bin/goveralls -service=travis-ci
@@ -0,0 +1,15 @@
+ISC License
+
+Copyright (c) 2012-2016 Dave Collins <dave@davec.name>
+
+Permission to use, copy, modify, and distribute this software for any
+purpose with or without fee is hereby granted, provided that the above
+copyright notice and this permission notice appear in all copies.
+
+THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
@@ -0,0 +1,25 @@
+mgo - MongoDB driver for Go
+
+Copyright (c) 2010-2013 - Gustavo Niemeyer <gustavo@niemeyer.net>
+
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met: 
+
+1. Redistributions of source code must retain the above copyright notice, this
+   list of conditions and the following disclaimer. 
+2. Redistributions in binary form must reproduce the above copyright notice,
+   this list of conditions and the following disclaimer in the documentation
+   and/or other materials provided with the distribution. 
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
@@ -0,0 +1,25 @@
+BSON library for Go
+
+Copyright (c) 2010-2012 - Gustavo Niemeyer <gustavo@niemeyer.net>
+
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met: 
+
+1. Redistributions of source code must retain the above copyright notice, this
+   list of conditions and the following disclaimer. 
+2. Redistributions in binary form must reproduce the above copyright notice,
+   this list of conditions and the following disclaimer in the documentation
+   and/or other materials provided with the distribution. 
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
@@ -0,0 +1,27 @@
+Copyright (c) 2012 The Go Authors. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+   * Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+   * Redistributions in binary form must reproduce the above
+copyright notice, this list of conditions and the following disclaimer
+in the documentation and/or other materials provided with the
+distribution.
+   * Neither the name of Google Inc. nor the names of its
+contributors may be used to endorse or promote products derived from
+this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
@@ -0,0 +1,27 @@
+Copyright (c) 2013, Patrick Mezard
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+    Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+    Redistributions in binary form must reproduce the above copyright
+notice, this list of conditions and the following disclaimer in the
+documentation and/or other materials provided with the distribution.
+    The names of its contributors may not be used to endorse or promote
+products derived from this software without specific prior written
+permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
+IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
+PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
+TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
@@ -0,0 +1,22 @@
+Copyright (c) 2012 - 2013 Mat Ryer and Tyler Bunnell
+
+Please consider promoting this project if you find it useful.
+
+Permission is hereby granted, free of charge, to any person 
+obtaining a copy of this software and associated documentation 
+files (the "Software"), to deal in the Software without restriction, 
+including without limitation the rights to use, copy, modify, merge, 
+publish, distribute, sublicense, and/or sell copies of the Software, 
+and to permit persons to whom the Software is furnished to do so, 
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included
+in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, 
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES 
+OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. 
+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, 
+DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT 
+OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE 
+OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2018 Umputun
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
@@ -0,0 +1,58 @@
+# Mongo [![Build Status](https://travis-ci.org/go-pkgz/mongo.svg?branch=master)](https://travis-ci.org/go-pkgz/mongo) [![Go Report Card](https://goreportcard.com/badge/github.com/go-pkgz/mongo)](https://goreportcard.com/report/github.com/go-pkgz/mongo) [![Coverage Status](https://coveralls.io/repos/github/go-pkgz/mongo/badge.svg?branch=master)](https://coveralls.io/github/go-pkgz/mongo?branch=master)
+
+Provides helpers on top of [mgo](https://github.com/globalsign/mgo)
+
+## Install and update
+
+`go get -u github.com/go-pkgz/mongo`
+
+
+## Usage
+
+- `Server` represents mongo instance and provides session accessor. Application usually creates one server object and uses it for anything needed with this particular mongo host or replica set.
+
+- `Connection` encapsulates session and provides auto-closable wrapper. Each requests runs inside one of With* function makes new mongo session and closes on completion.
+
+- `BufferedWriter` implements buffered writer to mongo. Write method caching internally till it reached buffer size. Flush methods can be called manually at any time.
+
+
+```golang
+    m, err := NewServerWithURL("mongodb://127.0.0.1:27017/test?debug=true", 3*time.Second)
+    if err != nil {
+        panic("can't make mongo server")
+    } 
+    
+    type testRecord struct {
+    	Key1 string
+    	Kay2 int
+    }
+    
+    err = c.WithCollection(func(coll *mgo.Collection) error { // create session
+        // insert 100 records
+        for i := 0; i < 100; i++ {
+            r := testRecord{
+                Key1: fmt.Sprintf("key-%02d", i%5),
+                Key2: i,
+            }
+            if e := coll.Insert(r); e != nil {
+                return e
+            }
+        }
+        return nil
+    })
+    
+```
+
+## Dependencies 
+
+- [globalsign/mgo](https://github.com/globalsign/mgo) - mgo mongo driver
+- [stretchr/testify/](https://github.com/stretchr/testify) - testing library (test-only dependency)
+
+## Testing
+
+`testing.go` helps to create test for real mongo (not mocks)
+
+- `mongo.MakeTestConnection` creates `mongo.Connection` for url defined in env `MONGO_TEST`. If not defined `mongodb://mongo:27017` used. By default it will use random connection with prefix `test_` in `test` DB.
+- `mongo.RemoveTestCollection` - drops collection used by `MakeTestConnection`
+- `mongo.RemoveTestCollections` - drops user-defined collections from `test` DB
+
@@ -0,0 +1,67 @@
+package mongo
+
+import (
+	"fmt"
+
+	"github.com/globalsign/mgo"
+)
+
+// sessionFn is a function for all With*Collection calls
+type sessionFn func(coll *mgo.Collection) error
+
+// Connection allows to run request in separate session, closing automatically
+type Connection struct {
+	server         *Server
+	db, collection string
+}
+
+// NewConnection makes a connection for server
+func NewConnection(server *Server, db string, collection string) *Connection {
+	return &Connection{server: server, db: db, collection: collection}
+}
+
+// WithCollection passes fun with mgo.Collection from session copy, closes it after done,
+// uses Connection.DB and Connection.Collection
+func (c *Connection) WithCollection(fun sessionFn) (err error) {
+	return c.WithCustomCollection(c.collection, fun)
+}
+
+// WithCustomCollection passes fun with mgo.Collection from session copy, closes it after done
+// uses Connection.DB or (if not defined) dial.Database, and user-defined collection
+func (c *Connection) WithCustomCollection(collection string, fun sessionFn) (err error) {
+	db := c.server.dial.Database
+	if c.db != "" {
+		db = c.db
+	}
+	return c.WithCustomDbCollection(db, collection, fun)
+}
+
+// WithCustomDbCollection passed fun with mgo.Collection from session copy, closes it after done
+// uses passed db and collection directly.
+func (c *Connection) WithCustomDbCollection(db string, collection string, fun sessionFn) (err error) {
+	session := c.server.SessionCopy()
+	defer session.Close()
+	return fun(session.DB(db).C(collection))
+}
+
+// WithDB passes fun with mgo.Database from session copy, closes it after done
+// uses Connection.DB or (if not defined) dial.Database
+func (c *Connection) WithDB(fun func(dbase *mgo.Database) error) (err error) {
+	db := c.server.dial.Database
+	if c.db != "" {
+		db = c.db
+	}
+	return c.WithCustomDB(db, fun)
+}
+
+// WithCustomDB passes fun with mgo.Database from session copy, closes it after done
+// uses passed db directly
+func (c *Connection) WithCustomDB(db string, fun func(dbase *mgo.Database) error) (err error) {
+	session := c.server.SessionCopy()
+	defer session.Close()
+	return fun(session.DB(db))
+}
+
+func (c *Connection) String() string {
+	return fmt.Sprintf("mongo:%s, db:%s, collection:%s", c.server, c.db, c.collection)
+}
@@ -0,0 +1,117 @@
+// Package mongo wraps mgo to provide easier way to construct mongo server (with auth).
+// Connection provides With* func wrappers to run query with session copy
+package mongo
+
+import (
+	"crypto/tls"
+	"errors"
+	"fmt"
+	"log"
+	"net"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/globalsign/mgo"
+)
+
+// Server represents mongo instance and provides session accessor
+type Server struct {
+	dial   mgo.DialInfo
+	params ServerParams
+	sess   *mgo.Session
+}
+
+// ServerParams optional set of parameters
+type ServerParams struct {
+	ConsistencyMode mgo.Mode
+	Delay           int  // initial delay to give mongo server some time to start, in case if mongo part of the same compose
+	Debug           bool // turn on mgo debug mode
+	SSL             bool
+}
+
+// NewServerWithURL makes mongo server from url like
+// mongodb://remark42:password@127.0.0.1t:27017/test?ssl=true&replicaSet=Cluster0-shard-0&authSource=admin
+func NewServerWithURL(url string, timeout time.Duration) (res *Server, err error) {
+	dial, params, err := parseURL(url, timeout)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create mongo server from url %s, %s", url, err)
+	}
+	return NewServer(dial, params)
+}
+
+// NewServer doing auth if passwd != "" and can delay to make sure local mongo is up
+func NewServer(dial mgo.DialInfo, params ServerParams) (res *Server, err error) {
+	result := Server{dial: dial, params: params}
+
+	if params.Debug {
+		mgo.SetDebug(true)
+		mgo.SetLogger(log.New(os.Stdout, "MGO ", log.Ldate|log.Ltime|log.Lmicroseconds))
+	}
+
+	if len(dial.Addrs) == 0 {
+		return nil, errors.New("missing mongo address")
+	}
+
+	if params.Delay > 0 {
+		log.Printf("[DEBUG] initial mongo delay=%d", params.Delay)
+		time.Sleep(time.Duration(params.Delay) * time.Second)
+	}
+
+	log.Printf("[DEBUG] dial mongo %s, ssl=%v", dial.Addrs, params.SSL)
+
+	if params.SSL {
+		tlsConfig := &tls.Config{}
+		dial.DialServer = func(addr *mgo.ServerAddr) (net.Conn, error) {
+			conn, e := tls.Dial("tcp", addr.String(), tlsConfig)
+			return conn, e
+		}
+	}
+
+	session, err := mgo.DialWithInfo(&dial)
+	if err != nil {
+		return nil, fmt.Errorf("can't connect to mongo, %v", err)
+	}
+	session.SetMode(params.ConsistencyMode, true)
+	session.SetSyncTimeout(30 * time.Second)
+	session.SetSocketTimeout(dial.Timeout)
+
+	if dial.Username != "" && dial.Password != "" {
+		creds := &mgo.Credential{Username: dial.Username, Password: dial.Password, Source: dial.Source}
+		log.Printf("[DEBUG] login to mongo, user=%s, db=%s", creds.Username, creds.Source)
+		if err = session.Login(creds); err != nil {
+			return nil, fmt.Errorf("can't login to mongo, %v", err)
+		}
+	}
+
+	result.sess = session
+	return &result, nil
+}
+
+// SessionCopy returns copy of main session. Client should close it
+func (m Server) SessionCopy() *mgo.Session {
+	return m.sess.Copy()
+}
+
+func (m Server) String() string {
+	return fmt.Sprintf("%v%s", m.dial.Addrs, m.dial.Database)
+}
+
+// parseURL extends mgo with debug option and extracts ssl flag to make ServerParams
+func parseURL(mongoURL string, connectTimeout time.Duration) (mgo.DialInfo, ServerParams, error) {
+	params := ServerParams{
+		ConsistencyMode: mgo.Monotonic,
+		SSL:             strings.Contains(mongoURL, "ssl=true"),
+		Debug:           strings.Contains(mongoURL, "debug=true"),
+	}
+
+	mongoURL = strings.Replace(mongoURL, "&debug=true", "", 1)
+	mongoURL = strings.Replace(mongoURL, "?debug=true", "", 1)
+
+	dial, err := mgo.ParseURL(mongoURL)
+	if err != nil {
+		return mgo.DialInfo{}, ServerParams{}, fmt.Errorf("failed to pars mongo url %s, %s", mongoURL, err)
+	}
+	dial.Timeout = connectTimeout
+	return *dial, params, nil
+}
@@ -0,0 +1,76 @@
+package mongo
+
+import (
+	"fmt"
+	"log"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/globalsign/mgo"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var conn *Connection
+var once sync.Once
+
+// MakeTestConnection connects to MONGO_TEST url or "mongo" host (in no env) and returns new connection.
+// collection name randomized on each call
+func MakeTestConnection(t *testing.T) (*Connection, error) {
+	mongoURL := getMongoURL(t)
+	once.Do(func() {
+		log.Print("[DEBUG] connect to mongo test instance")
+		srv, err := NewServerWithURL(mongoURL, 10*time.Second)
+		assert.Nil(t, err, "failed to dial")
+		collName := fmt.Sprintf("test_%d", time.Now().Nanosecond())
+		conn = NewConnection(srv, "test", collName)
+	})
+	RemoveTestCollection(t, conn)
+	return conn, nil
+}
+
+// RemoveTestCollection removes all records and drop collection from connection
+func RemoveTestCollection(t *testing.T, c *Connection) {
+	log.Printf("[DEBUG] clean test collection %+v", c.collection)
+	_ = c.WithCollection(func(coll *mgo.Collection) error {
+		_, e := coll.RemoveAll(nil)
+		require.Nil(t, e, "failed to remove records, %s", e)
+		e = coll.DropCollection()
+		if e != nil && e.Error() != "ns not found" {
+			require.Nil(t, e, "failed to drop collection, %s", e)
+		}
+		return e
+	})
+}
+
+// RemoveTestCollections clears passed collections
+func RemoveTestCollections(t *testing.T, c *Connection, collections ...string) {
+	log.Printf("[DEBUG] clean test collections %+v", collections)
+	for _, collection := range collections {
+		_ = c.WithCustomCollection(collection, func(coll *mgo.Collection) error {
+			_, e := coll.RemoveAll(nil)
+			require.Nil(t, e, "failed to remove records, %s", e)
+			e = coll.DropCollection()
+			if e != nil && e.Error() != "ns not found" {
+				require.Nil(t, e, "failed to drop collection, %s", e)
+			}
+			return e
+		})
+	}
+
+}
+
+func getMongoURL(t *testing.T) string {
+	mongoURL := os.Getenv("MONGO_TEST")
+	if mongoURL == "" {
+		mongoURL = "mongodb://mongo:27017"
+		t.Logf("no MONGO_TEST in env, defaulted to %s", mongoURL)
+	}
+	if mongoURL == "skip" {
+		t.Skip("skip mongo test")
+	}
+	return mongoURL
+}
@@ -0,0 +1,154 @@
+package mongo
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	"github.com/globalsign/mgo"
+)
+
+// BufferedWriter defines interface for writes and flush
+type BufferedWriter interface {
+	Write(rec interface{}) error
+	Flush() error
+	Close() error
+}
+
+// BufferedWriterMgo collects records in local buffer and flushes them as filled. Thread safe
+// by default using both DB and collection from provided connection.
+// Collection can be customized by WithCollection method. Optional flush duration to save on interval
+type BufferedWriterMgo struct {
+	connection    *Connection
+	bufferSize    int
+	collection    string
+	flushDuration time.Duration
+
+	ctx    context.Context
+	cancel context.CancelFunc
+
+	buffer        []interface{}
+	lock          sync.Mutex
+	lastWriteTime time.Time
+	once          sync.Once
+}
+
+// NewBufferedWriter makes batch writer for given size and connection
+func NewBufferedWriter(size int, connection *Connection) *BufferedWriterMgo {
+	if size == 0 {
+		size = 1
+	}
+	return &BufferedWriterMgo{
+		bufferSize: size,
+		buffer:     make([]interface{}, 0, size+1),
+		connection: connection,
+	}
+}
+
+// WithCollection sets custom collection to use with writer
+func (bw *BufferedWriterMgo) WithCollection(collection string) *BufferedWriterMgo {
+	bw.collection = collection
+	return bw
+}
+
+// WithAutoFlush sets auto flush duration
+func (bw *BufferedWriterMgo) WithAutoFlush(duration time.Duration) *BufferedWriterMgo {
+	bw.flushDuration = duration
+	if duration > 0 { // activate background auto-flush
+		bw.once.Do(func() {
+			bw.ctx, bw.cancel = context.WithCancel(context.Background())
+			ticker := time.NewTicker(duration)
+			go func() {
+				defer bw.cancel()
+				for {
+					select {
+					case <-ticker.C:
+						var shouldFlush bool
+						_ = bw.synced(func() error {
+							shouldFlush = time.Now().After(bw.lastWriteTime.Add(bw.flushDuration)) && len(bw.buffer) > 0
+							return nil
+						})
+						if shouldFlush {
+							if err := bw.Flush(); err != nil {
+								log.Printf("[WARN] flush failed, %s", err)
+							}
+						}
+					case <-bw.ctx.Done():
+						return
+					}
+				}
+			}()
+		})
+	}
+	return bw
+}
+
+// Write to buffer and, as filled, to mongo. If flushDuration defined check for automatic flush
+func (bw *BufferedWriterMgo) Write(rec interface{}) error {
+	return bw.synced(func() error {
+		bw.lastWriteTime = time.Now()
+		bw.buffer = append(bw.buffer, rec)
+		if len(bw.buffer) >= bw.bufferSize {
+			if err := bw.writeBuffer(); err != nil {
+				return fmt.Errorf("failed to write to %s, %s", bw.connection, err)
+			}
+			bw.buffer = bw.buffer[0:0]
+		}
+		return nil
+	})
+}
+
+// Flush writes everything left in buffer to mongo
+func (bw *BufferedWriterMgo) Flush() error {
+	err := bw.synced(func() error {
+		err := bw.writeBuffer()
+		bw.buffer = bw.buffer[0:0]
+		return err
+	})
+	if err != nil {
+		return fmt.Errorf("failed to flush to %s, %s", bw.connection, err)
+	}
+	return nil
+}
+
+// Close flushes all in-fly records and terminates background auto-flusher
+func (bw *BufferedWriterMgo) Close() (err error) {
+	return bw.synced(func() error {
+		err = bw.writeBuffer()
+		if bw.flushDuration > 0 {
+			bw.cancel()
+			<-bw.ctx.Done()
+		}
+		return err
+	})
+}
+
+// writeBuffer sends all collected records to mongo
+func (bw *BufferedWriterMgo) writeBuffer() (err error) {
+
+	if len(bw.buffer) == 0 {
+		return nil
+	}
+
+	if bw.collection == "" { // no custom collection
+		err = bw.connection.WithCollection(func(coll *mgo.Collection) error {
+			return coll.Insert(bw.buffer...)
+		})
+	}
+
+	if bw.collection != "" { // with custom collection
+		err = bw.connection.WithCustomCollection(bw.collection, func(coll *mgo.Collection) error {
+			return coll.Insert(bw.buffer...)
+		})
+	}
+
+	return err
+}
+
+func (bw *BufferedWriterMgo) synced(fn func() error) error {
+	bw.lock.Lock()
+	defer bw.lock.Unlock()
+	return fn()
+}