make create stream idempotent

src/connection.ts

@@ -21,6 +21,7 @@ import { CreateStreamResponse } from "./responses/create_stream_response"
 import { CreateStreamRequest, CreateStreamArguments } from "./requests/create_stream_request"
 import { Heartbeat } from "./heartbeat"
 import { TuneRequest } from "./requests/tune_request"
+import { STREAM_ALREADY_EXISTS_ERROR_CODE } from "./error_codes"
 
 export class Connection {
   private readonly socket = new Socket()
@@ -184,9 +185,13 @@ export class Connection {
   async createStream(params: { stream: string; arguments: CreateStreamArguments }): Promise<true> {
     this.logger.debug(`Create Stream...`)
     const res = await this.sendAndWait<CreateStreamResponse>(new CreateStreamRequest(params))
+    if (res.code === STREAM_ALREADY_EXISTS_ERROR_CODE) {
+      return true
+    }
     if (!res.ok) {
       throw new Error(`Create Stream command returned error with code ${res.code}`)
     }
+
     this.logger.debug(`Create Stream response: ${res.ok} - with arguments: '${inspect(params.arguments)}'`)
     return res.ok
   }

src/error_codes.ts

@@ -0,0 +1 @@
+export const STREAM_ALREADY_EXISTS_ERROR_CODE = 0x05

test/unit/create_stream.test.ts

@@ -47,21 +47,18 @@ describe("Stream", () => {
       expect(result.name).to.be.eql(streamName)
     })
 
-    it("Should detect a duplicate Stream", async () => {
+    it("Should be idempotent and ignore a duplicate Stream error", async () => {
       await connection.createStream({
         stream: streamName,
         arguments: payload,
       })
 
-      await expectToThrowAsync(
-        () =>
-          connection.createStream({
-            stream: streamName,
-            arguments: payload,
-          }),
-        Error,
-        "Create Stream command returned error with code 5"
-      )
+      const resp = await connection.createStream({
+        stream: streamName,
+        arguments: payload,
+      })
+
+      expect(resp).to.be.true
     })
 
     it("Should raise an error if creation goes wrong", async () => {