refactor(*): replace Observables with Promises

BREAKING CHANGE:
all requests no work with and return promises

Author: mumenthalers

package-lock.json

@@ -62,7 +62,6 @@
     "prettier": "^1.15.3",
     "reflect-metadata": "^0.1.12",
     "rimraf": "^2.6.3",
-    "rxjs": "^6.3.3",
     "semantic-release": "^16.0.0-beta.16",
     "sort-package-json": "^1.16.0",
     "ts-jest": "^23.10.4",
@@ -80,7 +79,6 @@
     "aws-sdk": "^2.382.0",
     "lodash": "^4.17.11",
     "reflect-metadata": "^0.1.12",
-    "rxjs": "^6.3.3",
     "uuid": "^3.3.2"
   }
 }

package.json

@@ -1,5 +1,4 @@
 import * as DynamoDB from 'aws-sdk/clients/dynamodb'
-import { of } from 'rxjs'
 import { DynamoRx } from '../dynamo-rx'
 import { batchGetItemsFetchAll, combineBatchGetResponses, hasUnprocessedKeys } from './batch-get-utils'
 
@@ -72,11 +71,11 @@ describe('batch-get utils', () => {
     }
 
     beforeEach(async () => {
-      batchGetItemsSpy = jasmine.createSpy().and.returnValues(of(output1), of(output2))
+      batchGetItemsSpy = jasmine.createSpy().and.returnValues(Promise.resolve(output1), Promise.resolve(output2))
       dynamoRx = <any>{ batchGetItems: batchGetItemsSpy }
       backoffTimerMock = { next: jasmine.createSpy().and.returnValue({ value: 0 }) }
 
-      await batchGetItemsFetchAll(dynamoRx, <any>{}, <IterableIterator<number>>(<any>backoffTimerMock), 0).toPromise()
+      await batchGetItemsFetchAll(dynamoRx, <any>{}, <IterableIterator<number>>(<any>backoffTimerMock), 0)
     })
 
     it('should use UnprocessedKeys for next request', () => {

src/dynamo/batchget/batch-get-utils.spec.ts

@@ -1,6 +1,5 @@
 import * as DynamoDB from 'aws-sdk/clients/dynamodb'
-import { Observable, of } from 'rxjs'
-import { delay, map, mergeMap } from 'rxjs/operators'
+import { promiseDelay } from '../../helper'
 import { DynamoRx } from '../dynamo-rx'
 
 /**
@@ -16,27 +15,25 @@ export function batchGetItemsFetchAll(
   params: DynamoDB.BatchGetItemInput,
   backoffTimer: IterableIterator<number>,
   throttleTimeSlot: number,
-): Observable<DynamoDB.BatchGetItemOutput> {
-  return dynamoRx.batchGetItems(params).pipe(
-    mergeMap(response => {
-      if (hasUnprocessedKeys(response)) {
-        // in case of unprocessedItems do a follow-up requests
-        return of(response.UnprocessedKeys).pipe(
+): Promise<DynamoDB.BatchGetItemOutput> {
+  return dynamoRx.batchGetItems(params)
+    .then(response => {
+        if (hasUnprocessedKeys(response)) {
+          // in case of unprocessedKeys do a follow-up requests
+          return Promise.resolve(response.UnprocessedKeys)
           // delay before doing the follow-up request
-          delay(backoffTimer.next().value * throttleTimeSlot),
-
-          mergeMap((UnprocessedKeys: DynamoDB.BatchGetRequestMap) => {
-            const nextParams = { ...params, RequestItems: UnprocessedKeys }
-            // call recursively batchGetItemsFetchAll with the returned UnprocessedItems params
-            return batchGetItemsFetchAll(dynamoRx, nextParams, backoffTimer, throttleTimeSlot)
-          }),
-          map(combineBatchGetResponses(response)),
-        )
-      }
-      // no follow-up request necessary, return result
-      return of(response)
-    }),
-  )
+            .then(promiseDelay(backoffTimer.next().value * throttleTimeSlot))
+            .then(UnprocessedKeys => {
+              const nextParams = { ...params, RequestItems: UnprocessedKeys }
+              // call recursively batchGetItemsFetchAll with the returned UnprocessedItems params
+              return batchGetItemsFetchAll(dynamoRx, nextParams, backoffTimer, throttleTimeSlot)
+            })
+            .then(combineBatchGetResponses(response))
+        }
+        // no follow-up request necessary, return result
+        return response
+      },
+    )
 }
 
 export type BatchGetItemOutputWithUnprocessedKeys = DynamoDB.BatchGetItemOutput & {

src/dynamo/batchget/batch-get-utils.ts

@@ -1,6 +1,5 @@
 // tslint:disable:no-non-null-assertion
 import * as DynamoDB from 'aws-sdk/clients/dynamodb'
-import { of } from 'rxjs'
 import { Organization, SimpleWithCompositePartitionKeyModel, SimpleWithPartitionKeyModel } from '../../../test/models'
 import { Attributes, toDb } from '../../mapper'
 import { DynamoRx } from '../dynamo-rx'
@@ -117,7 +116,7 @@ describe('batch get', () => {
       request = new BatchGetRequest()
       request.forModel(SimpleWithPartitionKeyModel, [jsItem1, jsItem2])
 
-      batchGetItemsSpy = jasmine.createSpy().and.returnValues(of(output1), of(output2))
+      batchGetItemsSpy = jasmine.createSpy().and.returnValues(Promise.resolve(output1), Promise.resolve(output2))
       const dynamoRx: DynamoRx = <any>{ batchGetItems: batchGetItemsSpy }
 
       Object.assign(request, { dynamoRx })
@@ -126,7 +125,7 @@ describe('batch get', () => {
     })
 
     it('[execNoMap] should backoff and retry when UnprocessedItems are returned', async () => {
-      const result = await request.execNoMap(generatorMock).toPromise()
+      const result = await request.execNoMap(generatorMock)
       expect(nextSpyFn).toHaveBeenCalledTimes(1)
       expect(batchGetItemsSpy).toHaveBeenCalledTimes(2)
       expect(result).toBeDefined()
@@ -140,7 +139,7 @@ describe('batch get', () => {
     })
 
     it('[execFullResponse] should backoff and retry when UnprocessedItems are returned', async () => {
-      const result = await request.execFullResponse(generatorMock).toPromise()
+      const result = await request.execFullResponse(generatorMock)
       expect(nextSpyFn).toHaveBeenCalledTimes(1)
       expect(batchGetItemsSpy).toHaveBeenCalledTimes(2)
       expect(result).toBeDefined()
@@ -154,7 +153,7 @@ describe('batch get', () => {
     })
 
     it('[exec] should backoff and retry when UnprocessedItems are returned', async () => {
-      const result = await request.exec(generatorMock).toPromise()
+      const result = await request.exec(generatorMock)
       expect(nextSpyFn).toHaveBeenCalledTimes(1)
       expect(batchGetItemsSpy).toHaveBeenCalledTimes(2)
       expect(result).toBeDefined()
@@ -182,21 +181,21 @@ describe('batch get', () => {
     }
 
     beforeEach(() => {
-      batchGetItemsSpy = jasmine.createSpy().and.returnValue(of(sampleResponse))
+      batchGetItemsSpy = jasmine.createSpy().and.returnValue(Promise.resolve(sampleResponse))
       const dynamoRx: DynamoRx = <any>{ batchGetItems: batchGetItemsSpy }
       request = new BatchGetRequest()
       Object.assign(request, { dynamoRx })
       request.forModel(SimpleWithPartitionKeyModel, [{ id: 'idVal' }])
     })
 
     it('exec', async () => {
-      const result = await request.exec().toPromise()
+      const result = await request.exec()
       expect(batchGetItemsSpy).toHaveBeenCalled()
       expect(result).toEqual({ [getTableName(SimpleWithPartitionKeyModel)]: [jsItem] })
     })
 
     it('execFullResponse', async () => {
-      const result = await request.execFullResponse().toPromise()
+      const result = await request.execFullResponse()
       expect(batchGetItemsSpy).toHaveBeenCalled()
       expect(result).toEqual({
         Responses: { [getTableName(SimpleWithPartitionKeyModel)]: [jsItem] },

src/dynamo/batchget/batch-get.request.spec.ts

@@ -1,6 +1,4 @@
 import * as DynamoDB from 'aws-sdk/clients/dynamodb'
-import { Observable } from 'rxjs'
-import { map } from 'rxjs/operators'
 import { metadataForClass } from '../../decorator/metadata/metadata-helper'
 import { randomExponentialBackoffTimer } from '../../helper'
 import { createToKeyFn, fromDb } from '../../mapper'
@@ -64,25 +62,25 @@ export class BatchGetRequest {
   execNoMap(
     backoffTimer = randomExponentialBackoffTimer,
     throttleTimeSlot = BATCH_GET_DEFAULT_TIME_SLOT,
-  ): Observable<DynamoDB.BatchGetItemOutput> {
+  ): Promise<DynamoDB.BatchGetItemOutput> {
     return this.fetch(backoffTimer, throttleTimeSlot)
   }
 
   execFullResponse(
     backoffTimer = randomExponentialBackoffTimer,
     throttleTimeSlot = BATCH_GET_DEFAULT_TIME_SLOT,
-  ): Observable<BatchGetFullResponse> {
-    return this.fetch(backoffTimer, throttleTimeSlot).pipe(map(this.mapResponse))
+  ): Promise<BatchGetFullResponse> {
+    return this.fetch(backoffTimer, throttleTimeSlot)
+      .then(this.mapResponse)
   }
 
   exec(
     backoffTimer = randomExponentialBackoffTimer,
     throttleTimeSlot = BATCH_GET_DEFAULT_TIME_SLOT,
-  ): Observable<BatchGetResponse> {
-    return this.fetch(backoffTimer, throttleTimeSlot).pipe(
-      map(this.mapResponse),
-      map(r => r.Responses),
-    )
+  ): Promise<BatchGetResponse> {
+    return this.fetch(backoffTimer, throttleTimeSlot)
+      .then(this.mapResponse)
+      .then(r => r.Responses)
   }
 
   private fetch(backoffTimer = randomExponentialBackoffTimer, throttleTimeSlot = BATCH_GET_DEFAULT_TIME_SLOT) {

src/dynamo/batchget/batch-get.request.ts

@@ -1,5 +1,4 @@
 import * as DynamoDB from 'aws-sdk/clients/dynamodb'
-import { of } from 'rxjs'
 import { DynamoRx } from '../dynamo-rx'
 import { batchWriteItemsWriteAll, hasUnprocessedItems } from './batch-write-utils'
 
@@ -21,11 +20,11 @@ describe('batch-write-utils', () => {
     const output2: DynamoDB.BatchWriteItemOutput = {}
 
     beforeEach(async () => {
-      batchWriteItemSpy = jasmine.createSpy().and.returnValues(of(output1), of(output2))
+      batchWriteItemSpy = jasmine.createSpy().and.returnValues(Promise.resolve(output1), Promise.resolve(output2))
       dynamoRx = <any>{ batchWriteItem: batchWriteItemSpy }
       backoffTimerMock = { next: jasmine.createSpy().and.returnValue({ value: 0 }) }
 
-      await batchWriteItemsWriteAll(dynamoRx, <any>{}, <IterableIterator<number>>(<any>backoffTimerMock), 0).toPromise()
+      await batchWriteItemsWriteAll(dynamoRx, <any>{}, <IterableIterator<number>>(<any>backoffTimerMock), 0)
     })
 
     it('should use UnprocessedKeys for next request', () => {

src/dynamo/batchwrite/batch-write-utils.spec.ts

@@ -1,6 +1,5 @@
 import * as DynamoDB from 'aws-sdk/clients/dynamodb'
-import { Observable, of } from 'rxjs'
-import { delay, mergeMap } from 'rxjs/operators'
+import { promiseDelay } from '../../helper'
 import { DynamoRx } from '../dynamo-rx'
 
 /**
@@ -16,27 +15,24 @@ export function batchWriteItemsWriteAll(
   params: DynamoDB.BatchWriteItemInput,
   backoffTimer: IterableIterator<number>,
   throttleTimeSlot: number,
-): Observable<DynamoDB.BatchGetItemOutput> {
-  return dynamoRx.batchWriteItem(params).pipe(
-    mergeMap(response => {
+): Promise<DynamoDB.BatchGetItemOutput> {
+  return dynamoRx.batchWriteItem(params)
+    .then(response => {
       if (hasUnprocessedItems(response)) {
         // in case of unprocessedItems do a follow-up requests
-        return of(response.UnprocessedItems).pipe(
+        return Promise.resolve(response.UnprocessedItems)
           // delay before doing the follow-up request
-          delay(backoffTimer.next().value * throttleTimeSlot),
-
-          mergeMap((unprocessedKeys: DynamoDB.BatchWriteItemRequestMap) => {
+          .then(promiseDelay(backoffTimer.next().value * throttleTimeSlot))
+          .then(unprocessedKeys => {
             const nextParams: DynamoDB.BatchWriteItemInput = { ...params, RequestItems: unprocessedKeys }
             // call recursively batchWriteItemsWriteAll with the returned UnprocessedItems params
             return batchWriteItemsWriteAll(dynamoRx, nextParams, backoffTimer, throttleTimeSlot)
-          }),
-          // no combining of responses necessary, only the last response is returned
-        )
+          })
+        // no combining of responses necessary, only the last response is returned
       }
       // no follow-up request necessary, return result
-      return of(response)
-    }),
-  )
+      return response
+    })
 }
 
 export type BatchWriteItemOutputWithUnprocessedItems = DynamoDB.BatchWriteItemOutput & {

src/dynamo/batchwrite/batch-write-utils.ts

@@ -1,4 +1,3 @@
-import { of } from 'rxjs'
 import { ComplexModel, SimpleWithPartitionKeyModel } from '../../../test/models'
 import { getTableName } from '../get-table-name.function'
 import { BatchWriteRequest } from './batch-write.request'
@@ -133,18 +132,18 @@ describe('batchWriteRequest', () => {
       const output = {
         myResponse: true,
       }
-      batchWriteItemSpy = jasmine.createSpy().and.returnValue(of(output))
+      batchWriteItemSpy = jasmine.createSpy().and.returnValue(Promise.resolve(output))
       const dynamoRx = <any>{ batchWriteItem: batchWriteItemSpy }
       req = new BatchWriteRequest()
       Object.assign(req, { dynamoRx })
     })
 
     it('exec should return void', async () => {
-      expect(await req.exec().toPromise()).toBeUndefined()
+      expect(await req.exec()).toBeUndefined()
     })
 
     it('execFullResponse should return the full response', async () => {
-      expect(await req.execFullResponse().toPromise()).toEqual({ myResponse: true })
+      expect(await req.execFullResponse()).toEqual({ myResponse: true })
     })
   })
 })

src/dynamo/batchwrite/batch-write.request.spec.ts

@@ -1,6 +1,4 @@
 import * as DynamoDB from 'aws-sdk/clients/dynamodb'
-import { Observable } from 'rxjs'
-import { map } from 'rxjs/operators'
 import { randomExponentialBackoffTimer } from '../../helper'
 import { createToKeyFn, toDb } from '../../mapper'
 import { ModelConstructor } from '../../model'
@@ -49,18 +47,15 @@ export class BatchWriteRequest {
   exec(
     backoffTimer = randomExponentialBackoffTimer,
     throttleTimeSlot = BATCH_WRITE_DEFAULT_TIME_SLOT,
-  ): Observable<void> {
-    return this.write(backoffTimer, throttleTimeSlot).pipe(
-      map(() => {
-        return
-      }),
-    )
+  ): Promise<void> {
+    return this.write(backoffTimer, throttleTimeSlot)
+      .then(() => { return })
   }
 
   execFullResponse(
     backoffTimer = randomExponentialBackoffTimer,
     throttleTimeSlot = BATCH_WRITE_DEFAULT_TIME_SLOT,
-  ): Observable<DynamoDB.BatchWriteItemOutput> {
+  ): Promise<DynamoDB.BatchWriteItemOutput> {
     return this.write(backoffTimer, throttleTimeSlot)
   }