Model penarikan umpan perubahan di Azure Cosmos DB

Untuk memproses umpan perubahan dengan menggunakan model penarikan, buat instans FeedIterator. Ketika Anda awalnya membuat FeedIterator, Anda harus menentukan nilai yang diperlukan ChangeFeedStartFrom , yang terdiri dari posisi awal untuk membaca perubahan dan nilai yang ingin Anda gunakan untuk FeedRange. FeedRange adalah rentang nilai kunci partisi dan menentukan item yang dapat dibaca dari umpan perubahan dengan menggunakan spesifik FeedIteratortersebut. Anda juga harus menentukan nilai yang diperlukan ChangeFeedMode untuk mode di mana Anda ingin memproses perubahan: versi terbaru atau semua versi dan penghapusan. Gunakan atau ChangeFeedMode.LatestVersion ChangeFeedMode.AllVersionsAndDeletes untuk menunjukkan mode mana yang ingin Anda gunakan untuk membaca umpan perubahan. Saat Anda menggunakan semua versi dan mode penghapusan, Anda harus memilih umpan perubahan mulai dari nilai baik Now() dari atau dari token kelanjutan tertentu.

Anda dapat secara opsional menentukan ChangeFeedRequestOptions untuk mengatur PageSizeHint. Saat diatur, properti ini menetapkan jumlah maksimum item yang diterima per halaman. Jika operasi dalam koleksi terpantau dilakukan melalui prosedur yang disimpan, cakupan transaksi dipertahankan saat membaca item dari umpan perubahan. Akibatnya, jumlah item yang diterima mungkin lebih tinggi dari nilai yang ditentukan sehingga item yang diubah oleh transaksi yang sama dikembalikan sebagai bagian dari satu batch atomik.

Berikut adalah contoh cara mendapatkan FeedIterator dalam mode versi terbaru yang mengembalikan objek entitas, dalam hal User ini objek:

FeedIterator<User> InteratorWithPOCOS = container.GetChangeFeedIterator<User>(ChangeFeedStartFrom.Beginning(), ChangeFeedMode.LatestVersion);

Semua versi dan mode penghapusan dalam pratinjau dan dapat digunakan dengan pratinjau versi >.NET SDK = 3.32.0-preview. Berikut adalah contoh untuk mendapatkan FeedIterator di semua versi dan menghapus mode yang mengembalikan User objek:

FeedIterator<ChangeFeedItem<User>> InteratorWithPOCOS = container.GetChangeFeedIterator<ChangeFeedItem<User>>(ChangeFeedStartFrom.Now(), ChangeFeedMode.AllVersionsAndDeletes);

Mengonsumsi umpan perubahan melalui aliran

FeedIterator untuk kedua mode umpan perubahan memiliki dua opsi. Selain contoh yang mengembalikan objek entitas, Anda juga dapat memperoleh respons dengan Stream dukungan. Aliran memungkinkan Anda membaca data tanpa terlebih dahulu deserialisasi, sehingga Anda menghemat sumber daya klien.

Berikut adalah contoh cara mendapatkan FeedIterator dalam mode versi terbaru yang mengembalikan Stream:

FeedIterator iteratorWithStreams = container.GetChangeFeedStreamIterator(ChangeFeedStartFrom.Beginning(), ChangeFeedMode.LatestVersion);

Mengonsumsi perubahan untuk seluruh kontainer

Jika Anda tidak menyediakan FeedRange parameter ke FeedIterator, Anda dapat memproses seluruh umpan perubahan kontainer dengan kecepatan Anda sendiri. Berikut adalah contoh, yang mulai membaca semua perubahan, dimulai pada waktu saat ini dengan menggunakan mode versi terbaru:

FeedIterator<User> iteratorForTheEntireContainer = container.GetChangeFeedIterator<User>(ChangeFeedStartFrom.Now(), ChangeFeedMode.LatestVersion);

while (iteratorForTheEntireContainer.HasMoreResults)
{
    FeedResponse<User> response = await iteratorForTheEntireContainer.ReadNextAsync();

    if (response.StatusCode == HttpStatusCode.NotModified)
    {
        Console.WriteLine($"No new changes");
        await Task.Delay(TimeSpan.FromSeconds(5));
    }
    else 
    {
        foreach (User user in response)
        {
            Console.WriteLine($"Detected change for user with id {user.id}");
        }
    }
}

Karena umpan perubahan secara efektif merupakan daftar item tak terbatas yang mencakup semua penulisan dan pembaruan di masa mendatang, nilainya HasMoreResults selalu true. Ketika Anda mencoba membaca umpan perubahan dan tidak ada perubahan baru yang tersedia, Anda menerima respons dengan NotModified status. Dalam contoh sebelumnya, ini ditangani dengan menunggu lima detik sebelum memeriksa ulang perubahan.

Mengonsumsi perubahan untuk kunci partisi

Dalam beberapa kasus, Anda mungkin hanya ingin memproses perubahan untuk kunci partisi tertentu. Anda dapat memperoleh FeedIterator kunci partisi tertentu dan memproses perubahan dengan cara yang sama seperti yang Anda bisa untuk seluruh kontainer.

FeedIterator<User> iteratorForPartitionKey = container.GetChangeFeedIterator<User>(
    ChangeFeedStartFrom.Beginning(FeedRange.FromPartitionKey(new PartitionKey("PartitionKeyValue")), ChangeFeedMode.LatestVersion));

while (iteratorForThePartitionKey.HasMoreResults)
{
    FeedResponse<User> response = await iteratorForThePartitionKey.ReadNextAsync();

    if (response.StatusCode == HttpStatusCode.NotModified)
    {
        Console.WriteLine($"No new changes");
        await Task.Delay(TimeSpan.FromSeconds(5));
    }
    else
    {
        foreach (User user in response)
        {
            Console.WriteLine($"Detected change for user with id {user.id}");
        }
    }
}

Gunakan FeedRange untuk paralelisasi

Dalam prosesor umpan perubahan, pekerjaan tersebar di beberapa konsumen secara otomatis. Dalam model penarikan umpan perubahan, Anda dapat menggunakan FeedRange untuk paralelisasi pemrosesan umpan perubahan. FeedRange mewakili rentang nilai kunci partisi.

Berikut adalah contoh yang menunjukkan cara mendapatkan daftar rentang untuk kontainer Anda:

IReadOnlyList<FeedRange> ranges = await container.GetFeedRangesAsync();

Saat Anda mendapatkan daftar FeedRange nilai untuk kontainer, Anda mendapatkan satu FeedRange per partisi fisik.

Dengan menggunakan FeedRange, Anda dapat membuat FeedIterator untuk menyejajarkan pemrosesan umpan perubahan di beberapa komputer atau utas. Berbeda dengan contoh sebelumnya yang menunjukkan cara mendapatkan FeedIterator untuk seluruh kontainer atau satu kunci partisi, Anda dapat menggunakan FeedRanges untuk mendapatkan beberapa FeedIterator yang dapat memproses umpan perubahan secara paralel.

Saat ingin menggunakan FeedRanges, Anda harus memiliki proses orkestrator yang mendapatkan FeedRanges dan mendistribusikannya ke mesin tersebut. Distribusi ini mungkin:

• Menggunakan FeedRange.ToJsonString dan mendistribusikan nilai untai (karakter) ini. Konsumen dapat menggunakan nilai ini dengan FeedRange.FromJsonString.
• Jika distribusi sedang berproses, lewati referensi objek FeedRange.

Berikut adalah sampel yang menunjukkan cara membaca dari awal umpan perubahan kontainer dengan menggunakan dua komputer terpisah hipotetis yang dibaca secara paralel:

Mesin 1:

FeedIterator<User> iteratorA = container.GetChangeFeedIterator<User>(ChangeFeedStartFrom.Beginning(ranges[0]), ChangeFeedMode.LatestVersion);
while (iteratorA.HasMoreResults)
{
    FeedResponse<User> response = await iteratorA.ReadNextAsync();

    if (response.StatusCode == HttpStatusCode.NotModified)
    {
        Console.WriteLine($"No new changes");
        await Task.Delay(TimeSpan.FromSeconds(5));
    }
    else
    {
        foreach (User user in response)
        {
            Console.WriteLine($"Detected change for user with id {user.id}");
        }
    }
}

Mesin 2:

FeedIterator<User> iteratorB = container.GetChangeFeedIterator<User>(ChangeFeedStartFrom.Beginning(ranges[1]), ChangeFeedMode.LatestVersion);
while (iteratorB.HasMoreResults)
{
    FeedResponse<User> response = await iteratorA.ReadNextAsync();

    if (response.StatusCode == HttpStatusCode.NotModified)
    {
        Console.WriteLine($"No new changes");
        await Task.Delay(TimeSpan.FromSeconds(5));
    }
    else
    {
        foreach (User user in response)
        {
            Console.WriteLine($"Detected change for user with id {user.id}");
        }
    }
}

Menyimpan token kelanjutan

Anda dapat menyimpan posisi FeedIterator dengan mendapatkan token kelanjutan. Token kelanjutan adalah nilai untai (karakter) yang melacak perubahan terakhir yang diproses FeedIterator Anda dan mengizinkan FeedIterator untuk melanjutkan pada titik ini nantinya. Token kelanjutan, jika ditentukan, lebih diutamakan dari waktu mulai dan dimulai dari nilai awal. Kode berikut membaca umpan perubahan sejak pembuatan kontainer. Setelah tidak ada lagi perubahan yang tersedia, kode tersebut akan menahan token kelanjutan sehingga konsumsi umpan perubahan dapat dilanjutkan nanti.

FeedIterator<User> iterator = container.GetChangeFeedIterator<User>(ChangeFeedStartFrom.Beginning(), ChangeFeedMode.LatestVersion);

string continuation = null;

while (iterator.HasMoreResults)
{
    FeedResponse<User> response = await iterator.ReadNextAsync();

    if (response.StatusCode == HttpStatusCode.NotModified)
    {
        Console.WriteLine($"No new changes");
        continuation = response.ContinuationToken;
        // Stop the consumption since there are no new changes
        break;
    }
    else
    {
        foreach (User user in response)
        {
            Console.WriteLine($"Detected change for user with id {user.id}");
        }
    }
}

// Some time later when I want to check changes again
FeedIterator<User> iteratorThatResumesFromLastPoint = container.GetChangeFeedIterator<User>(ChangeFeedStartFrom.ContinuationToken(continuation), ChangeFeedMode.LatestVersion);

Saat Anda menggunakan mode versi terbaru, token kelanjutan FeedIterator tidak pernah kedaluwarsa selama kontainer Azure Cosmos DB masih ada. Saat Anda menggunakan semua versi dan mode penghapusan, FeedIterator token kelanjutan valid selama perubahan terjadi dalam jendela retensi untuk pencadangan berkelanjutan.

Untuk memproses umpan perubahan dengan menggunakan model penarikan, buat instans Iterator<FeedResponse<JsonNode>> responseIterator. Saat membuat CosmosChangeFeedRequestOptions, Anda harus menentukan tempat untuk mulai membaca umpan perubahan dan meneruskan FeedRange parameter yang ingin Anda gunakan. FeedRange adalah rentang nilai kunci partisi yang menentukan item yang dapat dibaca dari umpan perubahan.

Jika Anda ingin membaca umpan perubahan di semua versi dan mode penghapusan, Anda juga harus menentukan allVersionsAndDeletes() kapan Anda membuat CosmosChangeFeedRequestOptions. Semua versi dan mode penghapusan tidak mendukung pemrosesan umpan perubahan dari awal atau dari titik waktu tertentu. Anda harus memproses perubahan dari sekarang atau dari token kelanjutan. Semua versi dan mode penghapusan dalam pratinjau dan tersedia dalam versi >Java SDK = 4.42.0.

Mengonsumsi perubahan untuk seluruh kontainer

Jika Anda menentukan FeedRange.forFullRange(), Anda dapat memproses umpan perubahan untuk seluruh kontainer dengan kecepatan Anda sendiri. Anda dapat secara opsional menentukan nilai dalam byPage(). Saat diatur, properti ini menetapkan jumlah maksimum item yang diterima per halaman.

Berikut adalah contoh cara mendapatkan responseIterator nilai dalam mode versi terbaru:

CosmosChangeFeedRequestOptions options = CosmosChangeFeedRequestOptions
        .createForProcessingFromBeginning(FeedRange.forFullRange());
Iterator<FeedResponse<JsonNode>> responseIterator = container
    .queryChangeFeed(options, JsonNode.class)
    .byPage()
    .toIterable()
    .iterator();

Berikut adalah contoh cara mendapatkan responseIterator di semua versi dan mode penghapusan:

CosmosChangeFeedRequestOptions options = CosmosChangeFeedRequestOptions
    .createForProcessingFromNow(FeedRange.forFullRange())
    .allVersionsAndDeletes();

Iterator<FeedResponse<JsonNode>> responseIterator = container
    .queryChangeFeed(options, JsonNode.class)
    .byPage()
    .toIterable()
    .iterator();

Kami kemudian dapat mengulangi hasilnya. Karena umpan perubahan secara efektif merupakan daftar item tak terbatas yang mencakup semua penulisan dan pembaruan di masa mendatang, nilai responseIterator.hasNext() selalu true. Berikut adalah contoh dalam mode versi terbaru, yang membaca semua perubahan, mulai dari awal. Setiap perulangan mempertahankan token kelanjutan setelah memproses semua peristiwa. Ini mengambil dari titik terakhir yang diproses dalam umpan perubahan dan ditangani dengan menggunakan createForProcessingFromContinuation:

int i = 0;
List<JsonNode> results;
while (responseIterator.hasNext()) {
    FeedResponse<JsonNode> response = responseIterator.next();
    results = response.getResults();
    logger.info("Got " + results.size() + " items(s)");

    // applying the continuation token
    // only after processing all events
    options = CosmosChangeFeedRequestOptions
            .createForProcessingFromContinuation(response.getContinuationToken());
    i++;
    if (i >= 5) {
        // artificially breaking out of loop - not required in a real app
        System.out.println("breaking....");
        break;
    }
}

Mengonsumsi perubahan kunci partisi

Dalam beberapa kasus, Anda mungkin hanya ingin memproses perubahan untuk kunci partisi tertentu. Anda dapat memproses perubahan untuk kunci partisi tertentu dengan cara yang sama seperti yang Anda bisa untuk seluruh kontainer. Berikut adalah contoh yang menggunakan mode versi terbaru:

options = CosmosChangeFeedRequestOptions
        .createForProcessingFromBeginning(FeedRange.forLogicalPartition(new PartitionKey(partitionKey)));

responseIterator = container
    .queryChangeFeed(options, JsonNode.class)
    .byPage()
    .toIterable()
    .iterator();

int pkIndex = 0;

while (responseIterator.hasNext()) {
    FeedResponse<JsonNode> response = responseIterator.next();
    results = response.getResults();
    logger.info("Got " + results.size() + " items(s) retrieved");

    // applying the continuation token
    // only after processing all events
    options = CosmosChangeFeedRequestOptions
        .createForProcessingFromContinuation(response.getContinuationToken());
    pkIndex++;
    if (pkIndex >= 5) {
        // artificially breaking out of loop
        System.out.println("breaking....");
        break;
    }
}

Gunakan FeedRange untuk paralelisasi

Dalam prosesor umpan perubahan, pekerjaan tersebar di beberapa konsumen secara otomatis. Dalam model penarikan umpan perubahan, Anda dapat menggunakan FeedRange untuk paralelisasi pemrosesan umpan perubahan. FeedRange mewakili rentang nilai kunci partisi.

Berikut adalah contoh yang menggunakan mode versi terbaru yang menunjukkan cara mendapatkan daftar rentang untuk kontainer Anda:

Mono<List<FeedRange>> feedranges = resources.container.getFeedRanges();
List<FeedRange> feedRangeList = feedranges.block();

Ketika Anda mendapatkan daftar FeedRanges untuk kontainer Anda, Anda mendapatkan satu FeedRange per partisi fisik.

Dengan menggunakan FeedRange, Anda dapat menyejajarkan pemrosesan umpan perubahan di beberapa komputer atau utas. Tidak seperti contoh sebelumnya yang menunjukkan cara memproses perubahan untuk seluruh kontainer atau satu kunci partisi, Anda dapat menggunakan FeedRanges untuk memproses feed perubahan secara paralel.

Saat ingin menggunakan FeedRanges, Anda harus memiliki proses orkestrator yang mendapatkan FeedRanges dan mendistribusikannya ke mesin tersebut. Distribusi ini mungkin:

• Menggunakan FeedRange.toString() dan mendistribusikan nilai untai (karakter) ini.
• Jika distribusi sedang berproses, lewati referensi objek FeedRange.

Berikut adalah sampel yang menggunakan mode versi terbaru. Ini menunjukkan cara membaca dari awal umpan perubahan kontainer dengan menggunakan dua komputer terpisah hipotetis yang dibaca secara paralel:

Mesin 1:

FeedRange range1 = feedRangeList.get(0);
options = CosmosChangeFeedRequestOptions
        .createForProcessingFromBeginning(range1);

int machine1index = 0;
responseIterator = container
    .queryChangeFeed(options, JsonNode.class)
    .byPage()
    .toIterable()
    .iterator();

while (responseIterator.hasNext()) {
    FeedResponse<JsonNode> response = responseIterator.next();
    results = response.getResults();
    logger.info("Got " + results.size() + " items(s) retrieved");

    // applying the continuation token
    // only after processing all events
    options = CosmosChangeFeedRequestOptions
        .createForProcessingFromContinuation(response.getContinuationToken());

    machine1index++;

    if (machine1index >= 5) {
        // artificially breaking out of loop - not required in a real app
        System.out.println("breaking....");
        break;
    }
}

Mesin 2:

FeedRange range2 = feedRangeList.get(1);
options = CosmosChangeFeedRequestOptions
        .createForProcessingFromBeginning(range2);

responseIterator = container
    .queryChangeFeed(options, JsonNode.class)
    .byPage()
    .toIterable()
    .iterator();

int machine2index = 0;

while (responseIterator.hasNext()) {
    FeedResponse<JsonNode> response = responseIterator.next();
    results = response.getResults();
    logger.info("Got " + results.size() + " items(s) retrieved");

    // applying the continuation token
    // only after processing all events
    options = CosmosChangeFeedRequestOptions
        .createForProcessingFromContinuation(response.getContinuationToken());

    machine2index++;
    if (machine2index >= 5) {
        // artificially breaking out of loop - not required in a real app
        System.out.println("breaking....");
        break;
    }
}

Untuk memproses umpan perubahan dengan menggunakan model penarikan, buat instans ChangeFeedPullModelIterator. Ketika Anda awalnya membuat ChangeFeedPullModelIterator, Anda harus menentukan nilai yang diperlukan changeFeedStartFrom di dalam ChangeFeedIteratorOptions yang terdiri dari posisi awal untuk membaca perubahan dan sumber daya (kunci partisi atau FeedRange) yang perubahannya akan diambil.

Anda dapat secara opsional menggunakan maxItemCount untuk ChangeFeedIteratorOptions mengatur jumlah maksimum item yang diterima per halaman. Berikut adalah contoh cara mendapatkan iterator dalam mode versi terbaru yang mengembalikan objek entitas:

const options = {
    changeFeedStartFrom: ChangeFeedStartFrom.Now()
};

const iterator = container.items.getChangeFeedIterator(options);

Mengonsumsi perubahan untuk seluruh kontainer

Jika Anda tidak menyediakan FeedRange parameter atau PartitionKey di dalam ChangeFeedStartFrom, Anda dapat memproses seluruh umpan perubahan kontainer dengan kecepatan Anda sendiri. Berikut adalah contoh, yang mulai membaca semua perubahan, dimulai pada waktu saat ini:

async function waitFor(milliseconds: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, milliseconds));
}

const options = {
      changeFeedStartFrom: ChangeFeedStartFrom.Beginning()
};

const iterator = container.items.getChangeFeedIterator(options);

let timeout = 0;

while(iterator.hasMoreResults) {
    const response = await iterator.readNext();
    if (response.statusCode === StatusCodes.NotModified) {
        timeout = 5000;
    } 
    else {
        console.log("Result found", response.result);
        timeout = 0;
    }
    await waitFor(timeout);
}

Karena umpan perubahan secara efektif merupakan daftar item tak terbatas yang mencakup semua penulisan dan pembaruan di masa mendatang, nilainya hasMoreResults selalu true. Ketika Anda mencoba membaca umpan perubahan dan tidak ada perubahan baru yang tersedia, Anda menerima respons dengan NotModified status. Dalam contoh sebelumnya, ini ditangani dengan menunggu lima detik sebelum memeriksa ulang perubahan.

Mengonsumsi perubahan untuk kunci partisi

Dalam beberapa kasus, Anda mungkin hanya ingin memproses perubahan untuk kunci partisi tertentu. Anda dapat memperoleh iterator untuk kunci partisi tertentu dan memproses perubahan dengan cara yang sama seperti yang Anda dapat untuk seluruh kontainer.

async function waitFor(milliseconds: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, milliseconds));
}

const options = {
      changeFeedStartFrom: ChangeFeedStartFrom.Beginning("partitionKeyValue")
};

const iterator = container.items.getChangeFeedIterator(options);

let timeout = 0;

while(iterator.hasMoreResults) {
    const response = await iterator.readNext();
    if (response.statusCode === StatusCodes.NotModified) {
        timeout = 5000;
    } 
    else {
        console.log("Result found", response.result);
        timeout = 0;
    }
    await waitFor(timeout);
}

Gunakan FeedRange untuk paralelisasi

Dalam model penarikan umpan perubahan, Anda dapat menggunakan FeedRange untuk paralelisasi pemrosesan umpan perubahan. FeedRange mewakili rentang nilai kunci partisi.

Berikut adalah contoh yang menunjukkan cara mendapatkan daftar rentang untuk kontainer Anda:

const ranges = await container.getFeedRanges();

Saat Anda mendapatkan daftar FeedRange nilai untuk kontainer, Anda mendapatkan satu FeedRange per partisi fisik.

Dengan menggunakan FeedRange, Anda dapat membuat iterator untuk menyejajarkan pemrosesan umpan perubahan di beberapa komputer atau utas. Tidak seperti contoh sebelumnya yang menunjukkan cara mendapatkan iterator changefeed untuk seluruh kontainer atau satu kunci partisi, Anda dapat menggunakan FeedRanges untuk mendapatkan beberapa iterator, yang dapat memproses umpan perubahan secara paralel.

Berikut adalah sampel yang menunjukkan cara membaca dari awal umpan perubahan kontainer dengan menggunakan dua komputer terpisah hipotetis yang dibaca secara paralel:

Mesin 1:

async function waitFor(milliseconds: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, milliseconds));
}

const options = {
      changeFeedStartFrom: ChangeFeedStartFrom.Beginning(ranges[0])
};

const iterator = container.items.getChangeFeedIterator(options);

let timeout = 0;

while(iterator.hasMoreResults) {
    const response = await iterator.readNext();
    if (response.statusCode === StatusCodes.NotModified) {
        timeout = 5000;
    } 
    else {
        console.log("Result found", response.result);
        timeout = 0;
    }
    await waitFor(timeout);
}

Mesin 2:

async function waitFor(milliseconds: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, milliseconds));
}

const options = {
      changeFeedStartFrom: ChangeFeedStartFrom.Beginning(ranges[1])
};

const iterator = container.items.getChangeFeedIterator(options);

let timeout = 0;

while(iterator.hasMoreResults) {
    const response = await iterator.readNext();
    if (response.statusCode === StatusCodes.NotModified) {
        timeout = 5000;
    } 
    else {
        console.log("Result found", response.result);
        timeout = 0;
    }
    await waitFor(timeout);
}

Menyimpan token kelanjutan

Anda dapat menyimpan posisi iterator Anda dengan mendapatkan token kelanjutan. Token kelanjutan adalah nilai string yang melacak perubahan iterator changefeed Anda yang terakhir diproses dan memungkinkan iterator untuk dilanjutkan pada titik ini nanti. Token kelanjutan, jika ditentukan, lebih diutamakan dari waktu mulai dan dimulai dari nilai awal. Kode berikut membaca umpan perubahan sejak pembuatan kontainer. Setelah tidak ada lagi perubahan yang tersedia, kode tersebut akan menahan token kelanjutan sehingga konsumsi umpan perubahan dapat dilanjutkan nanti.

const options = {
      changeFeedStartFrom: ChangeFeedStartFrom.Beginning()
};

const iterator = container.items.getChangeFeedIterator(options);

let timeout = 0;
let continuation = "";
while(iterator.hasMoreResults) {
    const response = await iterator.readNext();
    if (response.statusCode === StatusCodes.NotModified) {
        continuation = response.continuationToken;
        break;
    } 
    else {
        console.log("Result found", response.result);
    }
}

// For checking any new changes using the continuation token
const continuationOptions = {
    changeFeedStartFrom: ChangeFeedStartFrom(continuation)
}
const newIterator = container.items.getChangeFeedIterator(continuationOptions);

Token kelanjutan tidak pernah kedaluwarsa selama kontainer Azure Cosmos DB masih ada.

Gunakan AsinkronIterator

Anda dapat menggunakan JavaScript Async Iterator untuk mengambil changefeed. Berikut adalah contoh untuk menggunakan Iterator Asinkron.

async function waitFor(milliseconds: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, milliseconds));
}
const options = {
      changeFeedStartFrom: ChangeFeedStartFrom.Beginning()
};
let timeout = 0;

for await(const result of container.items.getChangeFeedIterator(options).getAsyncIterator()) {
    if (result.statusCode === StatusCodes.NotModified) {
      timeout = 5000;
    }
    else {
      console.log("Result found", result.result);
      timeout = 0;
    }
    await waitFor(timeout);
}