Print oplogs

argos.go

@@ -0,0 +1,31 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"os"
+
+	"github.com/mongodb/mongo-go-driver/core/connstring"
+	argos "github.com/simagix/argos/core"
+)
+
+// example: argos "mongodb://localhost:27017/prudential?replicaSet=replset" students '[{"$match": {"operationType": "update"}}]'
+func main() {
+	flag.Parse()
+	if len(flag.Args()) < 2 {
+		fmt.Println("usage: argos uri collection")
+		os.Exit(1)
+	}
+
+	connStr, err := connstring.Parse(flag.Arg(0))
+	if err != nil {
+		panic(err)
+	}
+
+	var pipeline string
+	if len(flag.Args()) >= 3 {
+		pipeline = flag.Arg(2)
+		fmt.Println("pipeline:", pipeline)
+	}
+	argos.PrintOpLogs(connStr, flag.Arg(1), pipeline)
+}

build.sh

@@ -0,0 +1,9 @@
+#! /bin/bash
+# Copyright 2018 Kuei-chun Chen. All rights reserved.
+
+export version="master-$(date "+%Y%m%d.%s")"
+mkdir -p build
+env GOOS=linux GOARCH=amd64 go build -ldflags "-X main.version=$version" -o build/argos-linux-x64 argos.go
+env GOOS=darwin GOARCH=amd64 go build -ldflags "-X main.version=$version" -o build/argos-osx-x64 argos.go
+env GOOS=windows GOARCH=amd64 go build -ldflags "-X main.version=$version" -o build/argos-win-x64.exe argos.go
+

core/bson.go

@@ -0,0 +1,4 @@
+package argos
+
+// M is a convenient alias for a map[string]interface{} map, useful for dealing with BSON in a native way. For instance:
+type M map[string]interface{}

core/changeStreams.go

@@ -0,0 +1,53 @@
+package argos
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+
+	"github.com/mongodb/mongo-go-driver/bson"
+	"github.com/mongodb/mongo-go-driver/core/connstring"
+	"github.com/mongodb/mongo-go-driver/mongo"
+)
+
+// PrintOpLogs prints oplogs in JSON format
+func PrintOpLogs(connStr connstring.ConnString, collname string, pipelineStr string) {
+	client, err := mongo.Connect(context.Background(), connStr.String(), nil)
+	if err != nil {
+		panic(err)
+	}
+
+	fmt.Println("Watching", connStr.Database+"."+collname)
+	db := client.Database(connStr.Database)
+	coll := db.Collection(collname)
+	ctx := context.Background()
+	var pipeline interface{}
+	if pipelineStr != "" {
+		pipeline, _ = bson.ParseExtJSONArray(pipelineStr)
+	}
+	fmt.Println("pipeline", pipeline)
+	cur, err := coll.Watch(ctx, pipeline)
+	if err != nil {
+		panic(err)
+	}
+	defer cur.Close(ctx)
+
+	for cur.Next(ctx) {
+		elem := bson.NewDocument()
+
+		if err := cur.Decode(elem); err != nil {
+			log.Fatal(err)
+		}
+
+		bytes := []byte(elem.ToExtJSON(true))
+		var v = M{}
+		json.Unmarshal(bytes, &v)
+		bytes, _ = json.MarshalIndent(v, "", "  ")
+		fmt.Println(string(bytes))
+	}
+
+	if err := cur.Err(); err != nil {
+		log.Fatal(err)
+	}
+}