Reagindo às mudanças do banco de dados com o MongoDB Change Streams e Go

1
package main
2
3
import (
4
   "context"
5
   "fmt"
6
   "os"
7
   "sync"
8
9
   "go.mongodb.org/mongo-driver/bson"
10
   "go.mongodb.org/mongo-driver/mongo"
11
   "go.mongodb.org/mongo-driver/mongo/options"
12
)
13
14
func main() {
15
   client, err := mongo.Connect(context.TODO(), options.Client().ApplyURI(os.Getenv("ATLAS_URI")))
16
   if err != nil {
17
      panic(err)
18
   }
19
   defer client.Disconnect(context.TODO())
20
21
   database := client.Database("quickstart")
22
   episodesCollection := database.Collection("episodes")
23
}

1
episodesStream, err := episodesCollection.Watch(context.TODO(), mongo.Pipeline{})
2
if err != nil {
3
   panic(err)
4
}

1
episodesStream, err := episodesCollection.Watch(context.TODO(), mongo.Pipeline{})
2
if err != nil {
3
   panic(err)
4
}
5
6
defer episodesStream.Close(context.TODO())
7
8
for episodesStream.Next(context.TODO()) {
9
   var data bson.M
10
   if err := episodesStream.Decode(&data); err != nil {
11
      panic(err)
12
   }
13
   fmt.Printf("%v\n", data)
14
}

1
map[_id:map[_data:825E4EFCB9000000012B022C0100296E5A1004D960EAE47DBE4DC8AC61034AE145240146645F696400645E3B38511C9D4400004117E80004] clusterTime:{1582234809 1} documentKey:map[_id:ObjectID("5e3b38511c9d
2
4400004117e8")] fullDocument:map[_id:ObjectID("5e3b38511c9d4400004117e8") description:The second episode duration:30 podcast:ObjectID("5e3b37e51c9d4400004117e6") title:Episode #3] ns:map[coll:episodes 
3
db:quickstart] operationType:replace]

1
package main
2
3
import (
4
   "context"
5
   "fmt"
6
   "os"
7
   "sync"
8
9
   "go.mongodb.org/mongo-driver/bson"
10
   "go.mongodb.org/mongo-driver/mongo"
11
   "go.mongodb.org/mongo-driver/mongo/options"
12
)
13
14
func iterateChangeStream(routineCtx context.Context, waitGroup sync.WaitGroup, stream *mongo.ChangeStream) {
15
   defer stream.Close(routineCtx)
16
   defer waitGroup.Done()
17
   for stream.Next(routineCtx) {
18
      var data bson.M
19
      if err := stream.Decode(&data); err != nil {
20
            panic(err)
21
      }
22
      fmt.Printf("%v\n", data)
23
   }
24
}
25
26
func main() {
27
   client, err := mongo.Connect(context.TODO(), options.Client().ApplyURI(os.Getenv("ATLAS_URI")))
28
   if err != nil {
29
      panic(err)
30
   }
31
   defer client.Disconnect(context.TODO())
32
33
   database := client.Database("quickstart")
34
   episodesCollection := database.Collection("episodes")
35
36
   var waitGroup sync.WaitGroup
37
38
   episodesStream, err := episodesCollection.Watch(context.TODO(), mongo.Pipeline{})
39
   if err != nil {
40
      panic(err)
41
   }
42
   waitGroup.Add(1)
43
   routineCtx, cancelFn := context.WithCancel(context.Background())
44
   go iterateChangeStream(routineCtx, waitGroup, episodesStream)
45
46
   waitGroup.Wait()
47
}

1
matchPipeline := bson.D{
2
   {
3
      "$match", bson.D{
4
            {"operationType", "insert"},
5
            {"fullDocument.duration", bson.D{
6
               {"$gt", 30},
7
            }},
8
      },
9
   },
10
}
11
12
episodesStream, err := episodesCollection.Watch(context.TODO(), mongo.Pipeline{matchPipeline})

1
map[_id:map[_data:825E4F03CF000000012B022C0100296E5A1004D960EAE47DBE4DC8AC61034AE145240146645F696400645E4F03A01C9D44000063CCBD0004] clusterTime:{1582236623 1} documentKey:map[_id:ObjectID("5e4f03a01c9d
2
44000063ccbd")] fullDocument:map[_id:ObjectID("5e4f03a01c9d44000063ccbd") description:a quick start into mongodb duration:35 podcast:1234 title:getting started with mongodb] ns:map[coll:episodes db:qui
3
ckstart] operationType:insert]