Tutorial de fluxos de alterações e triggers com o Node.js

1
function closeChangeStream(timeInMs = 60000, changeStream) {
2
    return new Promise((resolve) => {
3
        setTimeout(() => {
4
            console.log("Closing the change stream");
5
            resolve(changeStream.close());
6
        }, timeInMs)
7
    })
8
};

1
async function monitorListingsUsingEventEmitter(client, timeInMs = 60000, pipeline = []){  
2
3
}

1
const collection = client.db("sample_airbnb").collection("listingsAndReviews");

1
const changeStream = collection.watch(pipeline);

1
changeStream.on('change', (next) => {
2
     console.log(next);  
3
});

1
await closeChangeStream(timeInMs, changeStream);

1
await monitorListingsUsingEventEmitter(client);

1
{ 
2
    _id: { _data: '825DE67A42000000012B022C0100296E5A10046BBC1C6A9CBB4B6E9CA9447925E693EF46645F696400645DE67A42113EA7DE6472E7640004' },
3
    operationType: 'insert',
4
    clusterTime: Timestamp { _bsontype: 'Timestamp', low_: 1, high_: 1575385666 },
5
    fullDocument: { 
6
        _id: 5de67a42113ea7de6472e764,
7
        name: 'Opera House Views',
8
        summary: 'Beautiful apartment with views of the iconic Sydney Opera House',
9
        property_type: 'Apartment',
10
        bedrooms: 1,
11
        bathrooms: 1,
12
        beds: 1,
13
        address: { market: 'Sydney', country: 'Australia' } 
14
     },
15
     ns: { db: 'sample_airbnb', coll: 'listingsAndReviews' },
16
     documentKey: { _id: 5de67a42113ea7de6472e764 } 
17
}
18
{ 
19
    _id: { _data: '825DE67A42000000022B022C0100296E5A10046BBC1C6A9CBB4B6E9CA9447925E693EF46645F696400645DE67A42113EA7DE6472E7650004' },
20
    operationType: 'insert',
21
    clusterTime: Timestamp { _bsontype: 'Timestamp', low_: 2, high_: 1575385666 },
22
    fullDocument: { 
23
        _id: 5de67a42113ea7de6472e765,
24
        name: 'Private room in London',
25
        property_type: 'Apartment',
26
        bedrooms: 1,
27
        bathroom: 1 
28
    },
29
    ns: { db: 'sample_airbnb', coll: 'listingsAndReviews' },
30
    documentKey: { _id: 5de67a42113ea7de6472e765 } 
31
}
32
{ 
33
    _id: { _data: '825DE67A42000000032B022C0100296E5A10046BBC1C6A9CBB4B6E9CA9447925E693EF46645F696400645DE67A42113EA7DE6472E7660004' },
34
    operationType: 'insert',
35
    clusterTime: Timestamp { _bsontype: 'Timestamp', low_: 3, high_: 1575385666 },
36
    fullDocument: { 
37
        _id: 5de67a42113ea7de6472e766,
38
        name: 'Beautiful Beach House',
39
        summary: 'Enjoy relaxed beach living in this house with a private beach',
40
        bedrooms: 4,
41
        bathrooms: 2.5,
42
        beds: 7,
43
        last_review: 2019-12-03T15:07:46.730Z 
44
    },
45
    ns: { db: 'sample_airbnb', coll: 'listingsAndReviews' },
46
    documentKey: { _id: 5de67a42113ea7de6472e766 } 
47
 }
48
 { 
49
    _id: { _data: '825DE67A42000000042B022C0100296E5A10046BBC1C6A9CBB4B6E9CA9447925E693EF46645F696400645DE67A42113EA7DE6472E7640004' },
50
    operationType: 'update',
51
    clusterTime: Timestamp { _bsontype: 'Timestamp', low_: 4, high_: 1575385666 },
52
    ns: { db: 'sample_airbnb', coll: 'listingsAndReviews' },
53
    documentKey: { _id: 5de67a42113ea7de6472e764 },
54
    updateDescription: { 
55
        updatedFields: { beds: 2 }, 
56
        removedFields: [] 
57
    } 
58
 }
59
 { 
60
    _id: { _data: '825DE67A42000000052B022C0100296E5A10046BBC1C6A9CBB4B6E9CA9447925E693EF46645F696400645DE67A42113EA7DE6472E7660004' },
61
    operationType: 'update',
62
    clusterTime: Timestamp { _bsontype: 'Timestamp', low_: 5, high_: 1575385666 },
63
    ns: { db: 'sample_airbnb', coll: 'listingsAndReviews' },
64
    documentKey: { _id: 5de67a42113ea7de6472e766 },
65
    updateDescription: { 
66
        updatedFields: { address: [Object] }, 
67
        removedFields: [] 
68
    } 
69
 }
70
 { 
71
    _id: { _data: '825DE67A42000000062B022C0100296E5A10046BBC1C6A9CBB4B6E9CA9447925E693EF46645F696400645DE67A42113EA7DE6472E7670004' },
72
    operationType: 'insert',
73
    clusterTime: Timestamp { _bsontype: 'Timestamp', low_: 6, high_: 1575385666 },
74
    fullDocument: { 
75
        _id: 5de67a42113ea7de6472e767,
76
        name: 'Italian Villa',
77
        property_type: 'Entire home/apt',
78
        bedrooms: 6,
79
        bathrooms: 4,
80
        address: { market: 'Cinque Terre', country: 'Italy' } 
81
    },
82
    ns: { db: 'sample_airbnb', coll: 'listingsAndReviews' },
83
    documentKey: { _id: 5de67a42113ea7de6472e767 } 
84
 }
85
 { 
86
     _id: { _data: '825DE67A42000000072B022C0100296E5A10046BBC1C6A9CBB4B6E9CA9447925E693EF46645F696400645DE67A42113EA7DE6472E7680004' },
87
    operationType: 'insert',
88
    clusterTime: Timestamp { _bsontype: 'Timestamp', low_: 7, high_: 1575385666 },
89
    fullDocument: { 
90
        _id: 5de67a42113ea7de6472e768,
91
        name: 'Sydney Harbour Home',
92
        bedrooms: 4,
93
        bathrooms: 2.5,
94
        address: { market: 'Sydney', country: 'Australia' } },
95
        ns: { db: 'sample_airbnb', coll: 'listingsAndReviews' },
96
        documentKey: { _id: 5de67a42113ea7de6472e768 } 
97
 }
98
 { 
99
    _id: { _data: '825DE67A42000000082B022C0100296E5A10046BBC1C6A9CBB4B6E9CA9447925E693EF46645F696400645DE67A42113EA7DE6472E7680004' },
100
    operationType: 'delete',
101
    clusterTime: Timestamp { _bsontype: 'Timestamp', low_: 8, high_: 1575385666 },
102
    ns: { db: 'sample_airbnb', coll: 'listingsAndReviews' },
103
    documentKey: { _id: 5de67a42113ea7de6472e768 } 
104
 }

1
Closing the change stream

1
const pipeline = [
2
            {
3
                '$match': {
4
                    'operationType': 'insert',
5
                    'fullDocument.address.country': 'Australia',
6
                    'fullDocument.address.market': 'Sydney'
7
                },
8
            }
9
        ];

1
await monitorListingsUsingEventEmitter(client, 30000, pipeline);

1
{ 
2
   _id: { _data: '825DE67CED000000012B022C0100296E5A10046BBC1C6A9CBB4B6E9CA9447925E693EF46645F696400645DE67CED150EA2DF172344370004' },
3
   operationType: 'insert',
4
   clusterTime: Timestamp { _bsontype: 'Timestamp', low_: 1, high_: 1575386349 },
5
   fullDocument: { 
6
      _id: 5de67ced150ea2df17234437,
7
      name: 'Opera House Views',
8
      summary: 'Beautiful apartment with views of the iconic Sydney Opera House',
9
      property_type: 'Apartment',
10
      bedrooms: 1,
11
      bathrooms: 1,
12
      beds: 1,
13
      address: { market: 'Sydney', country: 'Australia' } 
14
   },
15
   ns: { db: 'sample_airbnb', coll: 'listingsAndReviews' },
16
   documentKey: { _id: 5de67ced150ea2df17234437 } 
17
}
18
{ 
19
   _id: { _data: '825DE67CEE000000032B022C0100296E5A10046BBC1C6A9CBB4B6E9CA9447925E693EF46645F696400645DE67CEE150EA2DF1723443B0004' },
20
   operationType: 'insert',
21
   clusterTime: Timestamp { _bsontype: 'Timestamp', low_: 3, high_: 1575386350 },
22
   fullDocument: { 
23
      _id: 5de67cee150ea2df1723443b,
24
      name: 'Sydney Harbour Home',
25
      bedrooms: 4,
26
      bathrooms: 2.5,
27
      address: { market: 'Sydney', country: 'Australia' } 
28
   },
29
   ns: { db: 'sample_airbnb', coll: 'listingsAndReviews' },
30
   documentKey: { _id: 5de67cee150ea2df1723443b } 
31
}

1
Closing the change stream

1
async function monitorListingsUsingHasNext(client, timeInMs = 60000, pipeline = []) {  
2
3
}

1
const collection = client.db("sample_airbnb").collection("listingsAndReviews");

1
const changeStream = collection.watch(pipeline);

1
closeChangeStream(timeInMs, changeStream);

1
try {
2
   while (await changeStream.hasNext()) {
3
      console.log(await changeStream.next());
4
   }
5
} catch (error) {
6
   if (changeStream.isClosed()) {
7
      console.log("The change stream is closed. Will not wait on any more changes.")
8
   } else {
9
      throw error;
10
  }
11
}

1
await monitorListingsUsingHasNext(client);

1
Closing the change stream

1
await monitorListingsUsingHasNext(client, 30000, pipeline);

1
Closing the change stream

1
const stream = require('stream');

1
async function monitorListingsUsingStreamAPI(client, timeInMs = 60000, pipeline = []) {  
2
3
}

1
const collection = client.db("sample_airbnb").collection("listingsAndReviews");

1
const changeStream = collection.watch(pipeline);

1
changeStream.stream().pipe(
2
      new stream.Writable({
3
          objectMode: true,
4
          write: function (doc, _, cb) {
5
              console.log(doc);
6
              cb();
7
          }
8
       })
9
);

1
await closeChangeStream(timeInMs, changeStream);

1
await monitorListingsUsingStreamAPI(client);

1
Closing the change stream

1
await monitorListingsUsingStreamAPI(client, 30000, pipeline);

1
Closing the change stream

1
exports = function(changeEvent) {
2
   console.log(JSON.stringify(changeEvent.fullDocument));  
3
};

1
{      
2
     "fullDocument.address.country": "Australia",      
3
     "fullDocument.address.market": "Sydney"  
4
}