Merge pull request #154 from shiftcode/#153-remove-rxjs

refactor(*): replace Observables with Promises

Author: michaelwittwer

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,6 +1,5 @@
 import * as DynamoDB from 'aws-sdk/clients/dynamodb'
-import { of } from 'rxjs'
-import { DynamoRx } from '../dynamo-rx'
+import { DynamoPromisified } from '../dynamo-promisified'
 import { batchGetItemsFetchAll, combineBatchGetResponses, hasUnprocessedKeys } from './batch-get-utils'
 
 describe('batch-get utils', () => {
@@ -52,7 +51,7 @@ describe('batch-get utils', () => {
 
   describe('batchGetItemsFetchAll', () => {
     let batchGetItemsSpy: jasmine.Spy
-    let dynamoRx: DynamoRx
+    let dynamoRx: DynamoPromisified
     let backoffTimerMock: { next: jasmine.Spy }
 
     const output1: DynamoDB.BatchGetItemOutput = {
@@ -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,7 +1,6 @@
 import * as DynamoDB from 'aws-sdk/clients/dynamodb'
-import { Observable, of } from 'rxjs'
-import { delay, map, mergeMap } from 'rxjs/operators'
-import { DynamoRx } from '../dynamo-rx'
+import { promiseDelay } from '../../helper'
+import { DynamoPromisified } from '../dynamo-promisified'
 
 /**
  * Function which executes batchGetItem operations until all given items (as params) are processed (fetched).
@@ -12,31 +11,29 @@ import { DynamoRx } from '../dynamo-rx'
  * @param throttleTimeSlot used to calculate the effective wait time
  */
 export function batchGetItemsFetchAll(
-  dynamoRx: DynamoRx,
+  dynamoRx: DynamoPromisified,
   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,9 +1,8 @@
 // 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'
+import { DynamoPromisified } from '../dynamo-promisified'
 import { getTableName } from '../get-table-name.function'
 import { BatchGetRequest } from './batch-get.request'
 
@@ -117,16 +116,16 @@ describe('batch get', () => {
       request = new BatchGetRequest()
       request.forModel(SimpleWithPartitionKeyModel, [jsItem1, jsItem2])
 
-      batchGetItemsSpy = jasmine.createSpy().and.returnValues(of(output1), of(output2))
-      const dynamoRx: DynamoRx = <any>{ batchGetItems: batchGetItemsSpy }
+      batchGetItemsSpy = jasmine.createSpy().and.returnValues(Promise.resolve(output1), Promise.resolve(output2))
+      const dynamoRx: DynamoPromisified = <any>{ batchGetItems: batchGetItemsSpy }
 
       Object.assign(request, { dynamoRx })
 
       nextSpyFn = jest.fn().mockImplementation(() => ({ value: 0 }))
     })
 
     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))
-      const dynamoRx: DynamoRx = <any>{ batchGetItems: batchGetItemsSpy }
+      batchGetItemsSpy = jasmine.createSpy().and.returnValue(Promise.resolve(sampleResponse))
+      const dynamoRx: DynamoPromisified = <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,12 +1,10 @@
 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'
 import { Attributes } from '../../mapper/type/attribute.type'
 import { ModelConstructor } from '../../model/model-constructor'
-import { DynamoRx } from '../dynamo-rx'
+import { DynamoPromisified } from '../dynamo-promisified'
 import { getTableName } from '../get-table-name.function'
 import { BatchGetFullResponse } from './batch-get-full.response'
 import { batchGetItemsFetchAll } from './batch-get-utils'
@@ -15,12 +13,12 @@ import { BatchGetResponse } from './batch-get.response'
 
 export class BatchGetRequest {
   readonly params: DynamoDB.BatchGetItemInput
-  private readonly dynamoRx: DynamoRx
+  private readonly dynamoRx: DynamoPromisified
   private readonly tables: Map<string, ModelConstructor<any>> = new Map()
   private itemCounter = 0
 
   constructor() {
-    this.dynamoRx = new DynamoRx()
+    this.dynamoRx = new DynamoPromisified()
     this.params = {
       RequestItems: {},
     }
@@ -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,12 +1,11 @@
 import * as DynamoDB from 'aws-sdk/clients/dynamodb'
-import { of } from 'rxjs'
-import { DynamoRx } from '../dynamo-rx'
+import { DynamoPromisified } from '../dynamo-promisified'
 import { batchWriteItemsWriteAll, hasUnprocessedItems } from './batch-write-utils'
 
 describe('batch-write-utils', () => {
   describe('batchWriteItemsWriteAll', () => {
     let batchWriteItemSpy: jasmine.Spy
-    let dynamoRx: DynamoRx
+    let dynamoRx: DynamoPromisified
     let backoffTimerMock: { next: jasmine.Spy }
 
     const output1: DynamoDB.BatchWriteItemOutput = {
@@ -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,7 +1,6 @@
 import * as DynamoDB from 'aws-sdk/clients/dynamodb'
-import { Observable, of } from 'rxjs'
-import { delay, mergeMap } from 'rxjs/operators'
-import { DynamoRx } from '../dynamo-rx'
+import { promiseDelay } from '../../helper'
+import { DynamoPromisified } from '../dynamo-promisified'
 
 /**
  * Function which executes batchWriteItem operations until all given items (as params) are processed (written).
@@ -12,31 +11,28 @@ import { DynamoRx } from '../dynamo-rx'
  * @param throttleTimeSlot used to calculate the effective wait time
  */
 export function batchWriteItemsWriteAll(
-  dynamoRx: DynamoRx,
+  dynamoRx: DynamoPromisified,
   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 })
     })
   })
 })