Absolutely! You can convert the Binary _data field from your change stream's resume token into a string for easy storage (in a database, file, etc.), then convert it back when you need to resume the stream. Here's a step-by-step guide with example code:

Step 1: Convert the Binary Resume Token to a String for Storage #

The Binary object in the Node.js MongoDB driver has a toString('base64') method that converts its buffer into a base64-encoded string—this is a safe, portable format for storing binary data as text.

const { MongoClient, Binary } = require('mongodb');
const fs = require('fs');

async function startAndTrackChangeStream() {
  // Connect to your MongoDB instance
  const client = await MongoClient.connect('mongodb://localhost:27017');
  const db = client.db('your-target-db');
  const collection = db.collection('your-target-collection');

  // Initialize the change stream
  const changeStream = collection.watch();

  console.log('Change stream started. Waiting for changes...');

  for await (const changeEvent of changeStream) {
    console.log('Detected change:', changeEvent);

    // Extract the resume token's binary data and convert to base64 string
    const resumeTokenString = changeEvent._id._data.toString('base64');

    // Option 1: Save to a local file
    fs.writeFileSync('resume-token.txt', resumeTokenString);
    console.log('Updated resume token saved to file');

    // Option 2: Save to a MongoDB collection for persistent storage
    await db.collection('resume-tokens').updateOne(
      { streamIdentifier: 'user-collection-stream' },
      { $set: { token: resumeTokenString, lastUpdated: new Date() } },
      { upsert: true }
    );
    console.log('Updated resume token saved to database');
  }
}

startAndTrackChangeStream().catch(err => console.error('Stream error:', err));

Step 2: Convert the String Back to a Binary Resume Token to Resume the Stream #

When you're ready to resume the change stream, retrieve the stored base64 string, convert it back to a Binary object using Binary.fromBase64(), and pass it to the resumeAfter option when initializing the stream.

async function resumeChangeStream() {
  const client = await MongoClient.connect('mongodb://localhost:27017');
  const db = client.db('your-target-db');
  const collection = db.collection('your-target-collection');

  // Retrieve the stored resume token string
  // Option 1: Read from local file
  const resumeTokenString = fs.readFileSync('resume-token.txt', 'utf8');

  // Option 2: Retrieve from MongoDB collection
  // const storedTokenDoc = await db.collection('resume-tokens').findOne({ streamIdentifier: 'user-collection-stream' });
  // const resumeTokenString = storedTokenDoc.token;

  // Convert base64 string back to Binary object
  const resumeTokenBinary = Binary.fromBase64(resumeTokenString);

  // Construct the resume token object (matches the structure MongoDB expects)
  const resumeToken = { _data: resumeTokenBinary };

  // Initialize the change stream with the resume token
  const changeStream = collection.watch([], { resumeAfter: resumeToken });

  console.log('Change stream resumed from stored token. Waiting for changes...');

  for await (const changeEvent of changeStream) {
    console.log('Detected change (resumed stream):', changeEvent);
    // Don't forget to update your stored resume token with the new one from each change event!
    const newResumeTokenString = changeEvent._id._data.toString('base64');
    fs.writeFileSync('resume-token.txt', newResumeTokenString);
  }
}

resumeChangeStream().catch(err => console.error('Resume error:', err));

Key Notes: #

• Base64 is ideal here: It encodes binary data into ASCII characters, so you won't run into encoding issues when storing or retrieving the string.
• Match the resume token structure: MongoDB expects the resume token to be an object with an _data field containing the Binary object—don't skip this structure, or the stream won't resume correctly.
• Update the token regularly: Always save the latest resume token from each change event, not just the first one. This ensures you can resume from the most recent point if the stream is interrupted.

内容的提问来源于stack exchange，提问作者tacticurv