Skip to content

Commit

Permalink
BREAKING: Change return type of the streams API to Record (#168)
Browse files Browse the repository at this point in the history
  • Loading branch information
uki00a authored Jan 3, 2021
1 parent be1a987 commit 0f3fe9b
Show file tree
Hide file tree
Showing 2 changed files with 40 additions and 34 deletions.
6 changes: 3 additions & 3 deletions stream.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ export interface XId {

export interface XMessage {
xid: XId;
field_values: Map<string, string>;
field_values: Record<string, string>;
}

export interface XKeyId {
Expand Down Expand Up @@ -215,15 +215,15 @@ export interface XClaimOpts {
export function parseXMessage(
raw: XReadIdData,
): XMessage {
const fieldValues: Map<string, string> = new Map();
const fieldValues: Record<string, string> = {};
let f: string | undefined = undefined;

let m = 0;
for (const data of raw[1]) {
if (m % 2 === 0) {
f = data;
} else if (f) {
fieldValues.set(f, data);
fieldValues[f] = data;
}
m++;
}
Expand Down
68 changes: 37 additions & 31 deletions tests/stream_test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -123,9 +123,10 @@ suite.test("xread", async () => {
{ air: "ball", friend: "table" },
{ elements: 10 },
);
const exampleMap = new Map<string, string>();
exampleMap.set("air", "horn");
exampleMap.set("friend", "fiend");
const exampleMap = {
air: "horn",
friend: "fiend",
};
const c = await client.xadd(key2, [1001, 1], exampleMap, { elements: 10 });
assert(c != null);

Expand All @@ -140,17 +141,20 @@ suite.test("xread", async () => {

assert(v != null);

const expectedAnimals = new Map();
expectedAnimals.set("cat", "moo");
expectedAnimals.set("dog", "honk");
expectedAnimals.set("duck", "yodel");

const expectedWeird = new Map();
expectedWeird.set("air", "ball");
expectedWeird.set("friend", "table");
const expectedOdd = new Map();
expectedOdd.set("air", "horn");
expectedOdd.set("friend", "fiend");
const expectedAnimals = {
cat: "moo",
dog: "honk",
duck: "yodel",
};

const expectedWeird = {
air: "ball",
friend: "table",
};
const expectedOdd = {
air: "horn",
friend: "fiend",
};
assertEquals(v, [
{
key,
Expand Down Expand Up @@ -246,7 +250,7 @@ suite.test("xreadgroup but no ack", async () => {
assertEquals(actualFirstStream.messages[0].xid, addedId);
assertEquals(actualFirstStream.messages.length, 1);
assertEquals(
actualFirstStream.messages[0].field_values.get("anyfield"),
actualFirstStream.messages[0].field_values["anyfield"],
"anyval",
);

Expand Down Expand Up @@ -298,9 +302,10 @@ suite.test("xadd with map then xread", async () => {

assert(v != null);

const expectedMap = new Map();
expectedMap.set("zoo", "theorize");
expectedMap.set("gable", "train");
const expectedMap = {
zoo: "theorize",
gable: "train",
};

assertEquals(v, [
{
Expand Down Expand Up @@ -335,9 +340,10 @@ suite.test("xadd with maxlen on map then xread", async () => {

assert(v != null);

const expectedMap = new Map();
expectedMap.set("hop", "4");
expectedMap.set("blip", "5");
const expectedMap = {
hop: "4",
blip: "5",
};

assertEquals(v, [
{ key, messages: [{ xid: addedId, field_values: expectedMap }] },
Expand Down Expand Up @@ -387,7 +393,7 @@ suite.test("unique message per consumer", async () => {

assertEquals(data[0].messages.length, 1);

assertEquals(data[0].messages[0].field_values.get("target"), payload);
assertEquals(data[0].messages[0].field_values["target"], payload);
}

await cleanupStream(client, key);
Expand Down Expand Up @@ -442,16 +448,16 @@ suite.test("xrange and xrevrange", async () => {
const basicResult = await client.xrange(key, "-", "+");
assertEquals(basicResult.length, 1);
assertEquals(basicResult[0].xid, firstId);
assertEquals(basicResult[0].field_values.get("f"), "v0");
assertEquals(basicResult[0].field_values["f"], "v0");

const secondId = await client.xadd(key, "*", { f: "v1" });
const revResult = await client.xrevrange(key, "+", "-");

assertEquals(revResult.length, 2);
assertEquals(revResult[0].xid, secondId);
assertEquals(revResult[0].field_values.get("f"), "v1");
assertEquals(revResult[0].field_values["f"], "v1");
assertEquals(revResult[1].xid, firstId);
assertEquals(revResult[1].field_values.get("f"), "v0");
assertEquals(revResult[1].field_values["f"], "v0");

// count should limit results
const lim = await client.xrange(key, "-", "+", 1);
Expand Down Expand Up @@ -508,11 +514,11 @@ suite.test("xclaim and xpending, all options", async () => {
assertEquals(firstClaimed.messages.length, 2);
assertEquals(
firstClaimed.messages[0].field_values,
new Map(Object.entries({ field: "foo" })),
{ field: "foo" },
);
assertEquals(
firstClaimed.messages[1].field_values,
new Map(Object.entries({ field: "bar" })),
{ field: "bar" },
);

// ACK these messages so we can try XPENDING/XCLAIM
Expand Down Expand Up @@ -630,11 +636,11 @@ suite.test("xclaim and xpending, all options", async () => {
assertEquals(thirdClaimed.messages.length, 2);
assertEquals(
thirdClaimed.messages[0].field_values,
new Map(Object.entries({ field: "woof", farm: "chicken" })),
{ field: "woof", farm: "chicken" },
);
assertEquals(
thirdClaimed.messages[1].field_values,
new Map(Object.entries({ field: "bop", farm: "duck" })),
{ field: "bop", farm: "duck" },
);
});
});
Expand All @@ -652,11 +658,11 @@ suite.test("xinfo", async () => {
assertEquals(basicStreamInfo.lastGeneratedId, { unixMs: 2, seqNo: 0 });
assertEquals(basicStreamInfo.firstEntry, {
xid: { unixMs: 1, seqNo: 0 },
field_values: new Map(Object.entries({ hello: "no" })),
field_values: { hello: "no" },
});
assertEquals(basicStreamInfo.lastEntry, {
xid: { unixMs: 2, seqNo: 0 },
field_values: new Map(Object.entries({ hello: "yes" })),
field_values: { hello: "yes" },
});

// Let's do an XREADGROUP so that we see some entries in the PEL
Expand Down

0 comments on commit 0f3fe9b

Please sign in to comment.