Skip to content

Commit

Permalink
Merge pull request #13 from keroxp/v0.3.0
Browse files Browse the repository at this point in the history
V0.3.0
  • Loading branch information
keroxp authored Jun 1, 2019
2 parents 897ab3b + 7e12bad commit 370e0a0
Show file tree
Hide file tree
Showing 9 changed files with 42 additions and 32 deletions.
4 changes: 2 additions & 2 deletions .travis.yml
Original file line number Diff line number Diff line change
@@ -1,11 +1,11 @@
language: python

install:
- curl -fsSL https://deno.land/x/install/install.sh | sh -s -- v0.3.3
- curl -fsSL https://deno.land/x/install/install.sh | sh -s -- v0.7.0
- export PATH="$HOME/.deno/bin:$PATH"

services:
- redis

script:
- make test
- deno run --allow-net test.ts
20 changes: 15 additions & 5 deletions io.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,8 @@
import { BufReader, BufWriter } from "https://deno.land/std@v0.3.2/io/bufio.ts";
import {
BufReader,
BufWriter,
EOF
} from "https://deno.land/std@v0.7.0/io/bufio.ts";
import Buffer = Deno.Buffer;
import { ErrorReplyError } from "./errors.ts";

Expand Down Expand Up @@ -38,7 +42,7 @@ export function createRequest(
export async function sendCommand(
writer: BufWriter,
reader: BufReader,
command,
command: string,
...args
): Promise<RedisRawReply> {
const msg = createRequest(command, ...args);
Expand All @@ -48,8 +52,11 @@ export async function sendCommand(
}

export async function readReply(reader: BufReader): Promise<RedisRawReply> {
const [b] = await reader.peek(1);
switch (b[0]) {
const res = await reader.peek(1);
if (res === EOF) {
throw EOF;
}
switch (res[0]) {
case IntegerReplyCode:
return ["integer", await readIntegerReply(reader)];
case SimpleStringCode:
Expand Down Expand Up @@ -118,7 +125,10 @@ export async function readArrayReply(reader: BufReader): Promise<any[]> {
const argCount = parseInt(line.substr(1, line.length - 3));
const result = [];
for (let i = 0; i < argCount; i++) {
const [res] = await reader.peek(1);
const res = await reader.peek(1);
if (res === EOF) {
throw EOF;
}
switch (res[0]) {
case SimpleStringCode:
result.push(await readStatusReply(reader));
Expand Down
2 changes: 1 addition & 1 deletion pipeline.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { BufReader, BufWriter } from "https://deno.land/std@v0.3.2/io/bufio.ts";
import { BufReader, BufWriter } from "https://deno.land/std@v0.7.0/io/bufio.ts";
import { createRequest, readReply, RedisRawReply } from "./io.ts";
import { ErrorReplyError } from "./errors.ts";
import { create, Redis } from "./redis.ts";
Expand Down
4 changes: 2 additions & 2 deletions pipeline_test.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import { test } from "https://deno.land/std@v0.3.2/testing/mod.ts";
import { assertEquals } from "https://deno.land/std@v0.3.2/testing/asserts.ts";
import { test } from "https://deno.land/std@v0.7.0/testing/mod.ts";
import { assertEquals } from "https://deno.land/std@v0.7.0/testing/asserts.ts";
import { connect } from "./redis.ts";

const addr = "127.0.0.1:6379";
Expand Down
18 changes: 9 additions & 9 deletions pubsub.ts
Original file line number Diff line number Diff line change
@@ -1,14 +1,14 @@
import { BufReader, BufWriter } from "https://deno.land/std@v0.3.2/io/bufio.ts";
import { readArrayReply, sendCommand } from "./io.ts";
import { BufReader, BufWriter } from "https://deno.land/std@v0.7.0/io/bufio.ts";
import { createRequest, readArrayReply, sendCommand } from "./io.ts";

export type RedisSubscription = {
readonly isClosed: boolean;
receive(): AsyncIterableIterator<RedisPubSubMessage>;
psubscribe(...patterns: string[]);
subscribe(...channels: string[]);
punsubscribe(...patterns: string[]);
unsubscribe(...channels: string[]);
close();
psubscribe(...patterns: string[]): Promise<void>;
subscribe(...channels: string[]): Promise<void>;
punsubscribe(...patterns: string[]): Promise<void>;
unsubscribe(...channels: string[]): Promise<void>;
close(): Promise<void>;
};

export type RedisPubSubMessage = {
Expand Down Expand Up @@ -89,7 +89,7 @@ export async function subscribe(
writer: BufWriter,
reader: BufReader,
...channels: string[]
) {
): Promise<RedisSubscription> {
const sub = new RedisSubscriptionImpl(writer, reader);
await sub.subscribe(...channels);
return sub;
Expand All @@ -99,7 +99,7 @@ export async function psubscribe(
writer: BufWriter,
reader: BufReader,
...patterns: string[]
) {
): Promise<RedisSubscription> {
const sub = new RedisSubscriptionImpl(writer, reader);
await sub.psubscribe(...patterns);
return sub;
Expand Down
18 changes: 9 additions & 9 deletions pubsub_test.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import { test } from "https://deno.land/std@v0.3.2/testing/mod.ts";
import { assertEquals } from "https://deno.land/std@v0.3.2/testing/asserts.ts";
import { test } from "https://deno.land/std@v0.7.0/testing/mod.ts";
import { assertEquals } from "https://deno.land/std@v0.7.0/testing/asserts.ts";
import { connect } from "./redis.ts";
import { RedisPubSubMessage } from "./pubsub.ts";

Expand All @@ -26,17 +26,17 @@ test(async function testSubscribe2() {
const pub = await connect(addr);
const sub = await redis.subscribe("subsc2");
let message: RedisPubSubMessage;
(async function() {
const p = (async function() {
const it = sub.receive();
message = (await it.next()).value;
})();
await pub.publish("subsc2", "wayway");
await sub.close();
await wait(100);
await p;
assertEquals(message, {
channel: "subsc2",
message: "wayway"
});
await sub.close();
const a = await redis.get("aaa");
assertEquals(a, void 0);
pub.close();
Expand All @@ -49,15 +49,14 @@ test(async function testPsubscribe() {
const sub = await redis.psubscribe("ps*");
let message1;
let message2;
(async function() {
const it = sub.receive();
const it = sub.receive();
const p = (async function() {
message1 = (await it.next()).value;
message2 = (await it.next()).value;
})();
await pub.publish("psub", "wayway");
await pub.publish("psubs", "heyhey");
await sub.close();
await wait(100);
await p;
assertEquals(message1, {
pattern: "ps*",
channel: "psub",
Expand All @@ -68,6 +67,7 @@ test(async function testPsubscribe() {
channel: "psubs",
message: "heyhey"
});
await sub.close();
pub.close();
redis.close();
});
2 changes: 1 addition & 1 deletion redis.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
type Reader = Deno.Reader;
type Writer = Deno.Writer;
type Closer = Deno.Closer;
import { BufReader, BufWriter } from "https://deno.land/std@v0.3.2/io/bufio.ts";
import { BufReader, BufWriter } from "https://deno.land/std@v0.7.0/io/bufio.ts";
import { ConnectionClosedError } from "./errors.ts";
import { psubscribe, RedisSubscription, subscribe } from "./pubsub.ts";
import { RedisRawReply, sendCommand } from "./io.ts";
Expand Down
4 changes: 2 additions & 2 deletions redis_test.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import { connect } from "./redis.ts";
import { test } from "https://deno.land/std@v0.3.2/testing/mod.ts";
import { assertEquals } from "https://deno.land/std@v0.3.2/testing/asserts.ts";
import { test } from "https://deno.land/std@v0.7.0/testing/mod.ts";
import { assertEquals } from "https://deno.land/std@v0.7.0/testing/asserts.ts";
// can be substituted with env variable
const addr = "127.0.0.1:6379";

Expand Down
2 changes: 1 addition & 1 deletion test.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import "./redis_test.ts";
import "./pubsub_test.ts";
import "./pipeline_test.ts";
import "https://deno.land/std@v0.3.2/testing/main.ts";
import "https://deno.land/std@v0.7.0/testing/main.ts";

0 comments on commit 370e0a0

Please sign in to comment.