#include <dataqueue/queue.h>

#include <gtest/gtest.h>

#include <node_bob-inl.h>

#include <util-inl.h>

#include <v8.h>

#include <memory>

#include <vector>

using node::DataQueue;

using v8::ArrayBuffer;

using v8::BackingStore;

TEST(DataQueue, InMemoryEntry) {

char buffer[] = "hello world";

size_t len = strlen(buffer);

std::shared_ptr<BackingStore> store = ArrayBuffer::NewBackingStore(

&buffer, len, [](void*, size_t, void*) {}, nullptr);

// We can create an InMemoryEntry from a v8::BackingStore.

std::unique_ptr<DataQueue::Entry> entry =

DataQueue::CreateInMemoryEntryFromBackingStore(store, 0, len);

// The entry is idempotent.

CHECK(entry->is_idempotent());

// The size is known.

CHECK_EQ(entry->size().value(), len);

// We can slice it.

// slice: "llo world"

std::unique_ptr<DataQueue::Entry> slice1 = entry->slice(2);

// The slice is idempotent.

CHECK(slice1->is_idempotent());

// The slice size is known.

CHECK_EQ(slice1->size().value(), len - 2);

// We can slice the slice with a length.

// slice: "o w"

uint64_t end = 5;

std::unique_ptr<DataQueue::Entry> slice2 = slice1->slice(2, end);

// That slice is idempotent.

CHECK(slice2->is_idempotent());

// That slice size is known.

CHECK_EQ(slice2->size().value(), 3);

// The slice end can extend beyond the actual size and will be adjusted.

// slice: "orld"

end = 100;

std::unique_ptr<DataQueue::Entry> slice3 = slice1->slice(5, end);

CHECK_NOT_NULL(slice3);

// The slice size is known.

CHECK_EQ(slice3->size().value(), 4);

// If the slice start is greater than the length, we get a zero length slice.

std::unique_ptr<DataQueue::Entry> slice4 = entry->slice(100);

CHECK_NOT_NULL(slice4);

CHECK_EQ(slice4->size().value(), 0);

// If the slice end is less than the start, we get a zero length slice.

end = 1;

std::unique_ptr<DataQueue::Entry> slice5 = entry->slice(2, end);

CHECK_NOT_NULL(slice5);

CHECK_EQ(slice5->size().value(), 0);

// If the slice end equal to the start, we get a zero length slice.

end = 2;

std::unique_ptr<DataQueue::Entry> slice6 = entry->slice(2, end);

CHECK_NOT_NULL(slice6);

CHECK_EQ(slice6->size().value(), 0);

// The shared_ptr for the BackingStore should show only 5 uses because

// the zero-length slices do not maintain a reference to it.

CHECK_EQ(store.use_count(), 5);

}

TEST(DataQueue, IdempotentDataQueue) {

char buffer1[] = "hello world";

char buffer2[] = "what fun this is";

char buffer3[] = "not added";

size_t len1 = strlen(buffer1);

size_t len2 = strlen(buffer2);

size_t len3 = strlen(buffer3);

std::shared_ptr<BackingStore> store1 = ArrayBuffer::NewBackingStore(

&buffer1, len1, [](void*, size_t, void*) {}, nullptr);

std::shared_ptr<BackingStore> store2 = ArrayBuffer::NewBackingStore(

&buffer2, len2, [](void*, size_t, void*) {}, nullptr);

std::vector<std::unique_ptr<DataQueue::Entry>> list;

list.push_back(

DataQueue::CreateInMemoryEntryFromBackingStore(store1, 0, len1));

list.push_back(

DataQueue::CreateInMemoryEntryFromBackingStore(store2, 0, len2));

// We can create an idempotent DataQueue from a list of entries.

std::shared_ptr<DataQueue> data_queue =

DataQueue::CreateIdempotent(std::move(list));

CHECK_NOT_NULL(data_queue);

// The data_queue is idempotent.

CHECK(data_queue->is_idempotent());

// The data_queue is capped.

CHECK(data_queue->is_capped());

// maybeCapRemaining() returns zero.

CHECK_EQ(data_queue->maybeCapRemaining().value(), 0);

// Calling cap() is a nonop but doesn't crash or error.

data_queue->cap();

data_queue->cap(100);

// maybeCapRemaining() still returns zero.

CHECK_EQ(data_queue->maybeCapRemaining().value(), 0);

// The size is known to be the sum of the in memory-entries.

CHECK_EQ(data_queue->size().value(), len1 + len2);

std::shared_ptr<BackingStore> store3 = ArrayBuffer::NewBackingStore(

&buffer3, len3, [](void*, size_t, void*) {}, nullptr);

// Trying to append a new entry does not crash, but returns std::nullopt.

CHECK(!data_queue

->append(DataQueue::CreateInMemoryEntryFromBackingStore(

store3, 0, len3))

.has_value());

// The size has not changed after the append.

CHECK_EQ(data_queue->size().value(), len1 + len2);

// We can acquire multiple readers from the data_queue.

std::shared_ptr<DataQueue::Reader> reader1 = data_queue->get_reader();

std::shared_ptr<DataQueue::Reader> reader2 = data_queue->get_reader();

CHECK_NOT_NULL(reader1);

CHECK_NOT_NULL(reader2);

const auto testRead = [&](auto& reader) {

// We can read the expected data from reader. Because the entries are

// InMemoryEntry instances, reads will be fully synchronous here.

bool waitingForPull = true;

// The first read produces buffer1

int status = reader->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

waitingForPull = false;

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

CHECK_EQ(count, 1);

CHECK_EQ(vecs[0].len, len1);

CHECK_EQ(memcmp(vecs[0].base, buffer1, len1), 0);

std::move(done)(0);

},

node::bob::OPTIONS_SYNC,

nullptr,

0,

node::bob::kMaxCountHint);

CHECK(!waitingForPull);

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

// We can read the expected data from reader1. Because the entries are

// InMemoryEntry instances, reads will be fully synchronous here.

waitingForPull = true;

// The second read should have status CONTINUE but no buffer.

status = reader->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

waitingForPull = false;

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

CHECK_EQ(count, 0);

},

node::bob::OPTIONS_SYNC,

nullptr,

0,

node::bob::kMaxCountHint);

CHECK(!waitingForPull);

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

// The third read produces buffer2, and should be the end.

status = reader->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

waitingForPull = false;

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

CHECK_EQ(count, 1);

CHECK_EQ(vecs[0].len, len2);

CHECK_EQ(memcmp(vecs[0].base, buffer2, len2), 0);

std::move(done)(0);

},

node::bob::OPTIONS_SYNC,

nullptr,

0,

node::bob::kMaxCountHint);

CHECK(!waitingForPull);

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

status = reader->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

waitingForPull = false;

CHECK_EQ(status, node::bob::STATUS_EOS);

CHECK_EQ(count, 0);

},

node::bob::OPTIONS_SYNC,

nullptr,

0,

node::bob::kMaxCountHint);

CHECK(!waitingForPull);

CHECK_EQ(status, node::bob::STATUS_EOS);

};

// Both reader1 and reader2 should pass identical tests.

testRead(reader1);

testRead(reader2);

// We can slice the data queue.

std::shared_ptr<DataQueue> slice1 = data_queue->slice(2);

CHECK_NOT_NULL(slice1);

// The slice is idempotent.

CHECK(slice1->is_idempotent());

// And capped.

CHECK(slice1->is_capped());

// The size is two-bytes less than the original.

CHECK_EQ(slice1->size().value(), data_queue->size().value() - 2);

const auto testSlice = [&](auto& reader) {

// We can read the expected data from reader. Because the entries are

// InMemoryEntry instances, reads will be fully synchronous here.

bool waitingForPull = true;

// The first read produces a slice of buffer1

int status = reader->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

waitingForPull = false;

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

CHECK_EQ(count, 1);

CHECK_EQ(vecs[0].len, len1 - 2);

CHECK_EQ(memcmp(vecs[0].base, buffer1 + 2, len1 - 2), 0);

std::move(done)(0);

},

node::bob::OPTIONS_SYNC,

nullptr,

0,

node::bob::kMaxCountHint);

CHECK(!waitingForPull);

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

// We can read the expected data from reader1. Because the entries are

// InMemoryEntry instances, reads will be fully synchronous here.

waitingForPull = true;

// The second read should have status CONTINUE but no buffer.

status = reader->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

waitingForPull = false;

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

CHECK_EQ(count, 0);

},

node::bob::OPTIONS_SYNC,

nullptr,

0,

node::bob::kMaxCountHint);

CHECK(!waitingForPull);

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

// The second read produces buffer2, and should be the end.

status = reader->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

waitingForPull = false;

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

CHECK_EQ(count, 1);

CHECK_EQ(vecs[0].len, len2);

CHECK_EQ(memcmp(vecs[0].base, buffer2, len2), 0);

std::move(done)(0);

},

node::bob::OPTIONS_SYNC,

nullptr,

0,

node::bob::kMaxCountHint);

CHECK(!waitingForPull);

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

// The third read produces EOS

status = reader->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

waitingForPull = false;

CHECK_EQ(status, node::bob::STATUS_EOS);

CHECK_EQ(count, 0);

CHECK_NULL(vecs);

std::move(done)(0);

},

node::bob::OPTIONS_SYNC,

nullptr,

0,

node::bob::kMaxCountHint);

CHECK(!waitingForPull);

CHECK_EQ(status, node::bob::STATUS_EOS);

};

// We can read the expected slice data.

std::shared_ptr<DataQueue::Reader> reader3 = slice1->get_reader();

testSlice(reader3);

// We can slice correctly across boundaries.

uint64_t end = 20;

std::shared_ptr<DataQueue> slice2 = data_queue->slice(5, end);

// The size is known.

CHECK_EQ(slice2->size().value(), 15);

const auto testSlice2 = [&](auto& reader) {

// We can read the expected data from reader. Because the entries are

// InMemoryEntry instances, reads will be fully synchronous here.

bool waitingForPull = true;

// The first read produces a slice of buffer1

int status = reader->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

waitingForPull = false;

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

CHECK_EQ(count, 1);

CHECK_EQ(vecs[0].len, len1 - 5);

CHECK_EQ(memcmp(vecs[0].base, buffer1 + 5, len1 - 5), 0);

std::move(done)(0);

},

node::bob::OPTIONS_SYNC,

nullptr,

0,

node::bob::kMaxCountHint);

CHECK(!waitingForPull);

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

// We can read the expected data from reader1. Because the entries are

// InMemoryEntry instances, reads will be fully synchronous here.

waitingForPull = true;

status = reader->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

waitingForPull = false;

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

CHECK_EQ(count, 0);

},

node::bob::OPTIONS_SYNC,

nullptr,

0,

node::bob::kMaxCountHint);

CHECK(!waitingForPull);

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

// The next read produces buffer2, and should be the end.

status = reader->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

waitingForPull = false;

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

CHECK_EQ(count, 1);

CHECK_EQ(vecs[0].len, len2 - 7);

CHECK_EQ(memcmp(vecs[0].base, buffer2, len2 - 7), 0);

std::move(done)(0);

},

node::bob::OPTIONS_SYNC,

nullptr,

0,

node::bob::kMaxCountHint);

CHECK(!waitingForPull);

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

// The next read produces EOS

status = reader->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

waitingForPull = false;

CHECK_EQ(status, node::bob::STATUS_EOS);

CHECK_EQ(count, 0);

CHECK_NULL(vecs);

std::move(done)(0);

},

node::bob::OPTIONS_SYNC,

nullptr,

0,

node::bob::kMaxCountHint);

CHECK(!waitingForPull);

CHECK_EQ(status, node::bob::STATUS_EOS);

};

// We can read the expected slice data.

std::shared_ptr<DataQueue::Reader> reader4 = slice2->get_reader();

testSlice2(reader4);

}

TEST(DataQueue, NonIdempotentDataQueue) {

char buffer1[] = "hello world";

char buffer2[] = "what fun this is";

char buffer3[] = "not added";

size_t len1 = strlen(buffer1);

size_t len2 = strlen(buffer2);

size_t len3 = strlen(buffer3);

std::shared_ptr<BackingStore> store1 = ArrayBuffer::NewBackingStore(

&buffer1, len1, [](void*, size_t, void*) {}, nullptr);

std::shared_ptr<BackingStore> store2 = ArrayBuffer::NewBackingStore(

&buffer2, len2, [](void*, size_t, void*) {}, nullptr);

std::shared_ptr<BackingStore> store3 = ArrayBuffer::NewBackingStore(

&buffer3, len3, [](void*, size_t, void*) {}, nullptr);

// We can create an non-idempotent DataQueue from a list of entries.

std::shared_ptr<DataQueue> data_queue = DataQueue::Create();

CHECK(!data_queue->is_idempotent());

CHECK_EQ(data_queue->size().value(), 0);

data_queue->append(

DataQueue::CreateInMemoryEntryFromBackingStore(store1, 0, len1));

CHECK_EQ(data_queue->size().value(), len1);

data_queue->append(

DataQueue::CreateInMemoryEntryFromBackingStore(store2, 0, len2));

CHECK_EQ(data_queue->size().value(), len1 + len2);

CHECK(!data_queue->is_capped());

CHECK(!data_queue->maybeCapRemaining().has_value());

data_queue->cap(100);

CHECK(data_queue->is_capped());

CHECK_EQ(data_queue->maybeCapRemaining().value(), 100 - (len1 + len2));

data_queue->cap(101);

CHECK(data_queue->is_capped());

CHECK_EQ(data_queue->maybeCapRemaining().value(), 100 - (len1 + len2));

data_queue->cap();

CHECK(data_queue->is_capped());

CHECK_EQ(data_queue->maybeCapRemaining().value(), 0);

// We can't add any more because the data queue is capped.

CHECK_EQ(data_queue

->append(DataQueue::CreateInMemoryEntryFromBackingStore(

store3, 0, len3))

.value(),

false);

// We cannot slice a non-idempotent data queue

std::shared_ptr<DataQueue> slice1 = data_queue->slice(2);

CHECK_NULL(slice1);

// We can acquire only a single reader for a non-idempotent data queue

std::shared_ptr<DataQueue::Reader> reader1 = data_queue->get_reader();

std::shared_ptr<DataQueue::Reader> reader2 = data_queue->get_reader();

CHECK_NOT_NULL(reader1);

CHECK_NULL(reader2);

const auto testRead = [&](auto& reader) {

// We can read the expected data from reader. Because the entries are

// InMemoryEntry instances, reads will be fully synchronous here.

bool waitingForPull = true;

// The first read produces buffer1

int status = reader->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

waitingForPull = false;

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

CHECK_EQ(count, 1);

CHECK_EQ(vecs[0].len, len1);

CHECK_EQ(memcmp(vecs[0].base, buffer1, len1), 0);

std::move(done)(0);

},

node::bob::OPTIONS_SYNC,

nullptr,

0,

node::bob::kMaxCountHint);

CHECK(!waitingForPull);

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

// The next read produces buffer2. When the first entry's reader returns

// EOS, the NonIdempotentDataQueueReader immediately pulls from the next

// entry (recursive Pull), so the transition is seamless.

waitingForPull = true;

status = reader->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

waitingForPull = false;

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

CHECK_EQ(count, 1);

CHECK_EQ(vecs[0].len, len2);

CHECK_EQ(memcmp(vecs[0].base, buffer2, len2), 0);

std::move(done)(0);

},

node::bob::OPTIONS_SYNC,

nullptr,

0,

node::bob::kMaxCountHint);

CHECK(!waitingForPull);

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

// The next read produces EOS

status = reader->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

waitingForPull = false;

CHECK_EQ(status, node::bob::STATUS_EOS);

CHECK_EQ(count, 0);

CHECK_NULL(vecs);

std::move(done)(0);

},

node::bob::OPTIONS_SYNC,

nullptr,

0,

node::bob::kMaxCountHint);

CHECK(!waitingForPull);

CHECK_EQ(status, node::bob::STATUS_EOS);

};

// Reading produces the expected results.

testRead(reader1);

// We still cannot acquire another reader.

std::shared_ptr<DataQueue::Reader> reader3 = data_queue->get_reader();

CHECK_NULL(reader3);

CHECK_NOT_NULL(data_queue);

}

TEST(DataQueue, DataQueueEntry) {

char buffer1[] = "hello world";

char buffer2[] = "what fun this is";

size_t len1 = strlen(buffer1);

size_t len2 = strlen(buffer2);

std::shared_ptr<BackingStore> store1 = ArrayBuffer::NewBackingStore(

&buffer1, len1, [](void*, size_t, void*) {}, nullptr);

std::shared_ptr<BackingStore> store2 = ArrayBuffer::NewBackingStore(

&buffer2, len2, [](void*, size_t, void*) {}, nullptr);

std::vector<std::unique_ptr<DataQueue::Entry>> list;

list.push_back(

DataQueue::CreateInMemoryEntryFromBackingStore(store1, 0, len1));

list.push_back(

DataQueue::CreateInMemoryEntryFromBackingStore(store2, 0, len2));

// We can create an idempotent DataQueue from a list of entries.

std::shared_ptr<DataQueue> data_queue =

DataQueue::CreateIdempotent(std::move(list));

CHECK_NOT_NULL(data_queue);

// We can create an Entry from a data queue.

std::unique_ptr<DataQueue::Entry> entry =

DataQueue::CreateDataQueueEntry(data_queue);

// The entry should be idempotent since the data queue is idempotent.

CHECK(entry->is_idempotent());

// The entry size should match the data queue size.

CHECK_EQ(entry->size().value(), data_queue->size().value());

// We can slice it since it is idempotent.

uint64_t end = 20;

std::unique_ptr<DataQueue::Entry> slice = entry->slice(5, end);

// The slice has the expected length.

CHECK_EQ(slice->size().value(), 15);

// We can add it to another data queue, even if the new one is not

// idempotent.

std::shared_ptr<DataQueue> data_queue2 = DataQueue::Create();

CHECK(data_queue2->append(std::move(slice)).value());

// Our original data queue should have a use count of 2.

CHECK_EQ(data_queue.use_count(), 2);

std::shared_ptr<DataQueue::Reader> reader = data_queue2->get_reader();

bool pullIsPending = true;

int status = reader->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

pullIsPending = false;

CHECK_EQ(count, 1);

CHECK_EQ(memcmp(vecs[0].base, buffer1 + 5, len1 - 5), 0);

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

},

node::bob::OPTIONS_SYNC,

nullptr,

0);

// All of the actual entries are in-memory entries so reads should be sync.

CHECK(!pullIsPending);

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

// Cap the queue so the reader can reach EOS after draining all entries.

data_queue2->cap();

// Read to completion...

while (status != node::bob::STATUS_EOS) {

status = reader->Pull(

[&](auto, auto, auto, auto) {}, node::bob::OPTIONS_SYNC, nullptr, 0);

}

// Because the original data queue is idempotent, we can still read from it,

// even though we have already consumed the non-idempotent data queue that

// contained it.

std::shared_ptr<DataQueue::Reader> reader2 = data_queue->get_reader();

CHECK_NOT_NULL(reader2);

pullIsPending = true;

status = reader2->Pull(

[&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {

pullIsPending = false;

CHECK_EQ(count, 1);

CHECK_EQ(memcmp(vecs[0].base, buffer1, len1), 0);

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

},

node::bob::OPTIONS_SYNC,

nullptr,

0);

// All of the actual entries are in-memory entries so reads should be sync.

CHECK(!pullIsPending);

CHECK_EQ(status, node::bob::STATUS_CONTINUE);

}