Merge pull request #53 from shiftcode/many-requests

michaelwittwer · web-flow · commit 5015542ff9eb · 2018-10-25T13:19:27.000+02:00
Many requests
diff --git a/src/dynamo/dynamo-store.ts b/src/dynamo/dynamo-store.ts
@@ -7,6 +7,7 @@ import { DEFAULT_TABLE_NAME_RESOLVER } from './default-table-name-resolver.const
 import { DynamoApiOperations } from './dynamo-api-operations.type'
 import { DynamoRx } from './dynamo-rx'
 import { BatchGetSingleTableRequest } from './request/batchgetsingletable/batch-get-single-table.request'
+import { BatchWriteSingleTableRequest } from './request/batchwritesingletable/batch-write-single-table.request'
 import { DeleteRequest } from './request/delete/delete.request'
 import { GetRequest } from './request/get/get.request'
 import { PutRequest } from './request/put/put.request'
@@ -58,6 +59,15 @@ export class DynamoStore<T> {
     return new DeleteRequest(this.dynamoRx, this.modelClazz, this.tableName, partitionKey, sortKey)
   }
 
+  /**
+   * This is a special implementation of batchWriteItem request, because it only supports one table,
+   * if you wish to write items to multiple tables
+   * create an instance of BatchWriteItemInput and use store.makeRequest with it.
+   */
+  batchWrite(): BatchWriteSingleTableRequest<T> {
+    return new BatchWriteSingleTableRequest<T>(this.dynamoRx, this.modelClazz, this.tableName)
+  }
+
   scan(): ScanRequest<T> {
     return new ScanRequest<T>(this.dynamoRx, this.modelClazz, this.tableName)
   }
@@ -67,8 +77,9 @@ export class DynamoStore<T> {
   }
 
   /**
-   * This is a special implementation of BatchGetItem request, because it only supports one table, if you wish to retrieve items from multiple tables
-   * get an instance of BatchGetItem request and call it there.
+   * This is a special implementation of BatchGetItem request, because it only supports one table,
+   * if you wish to retrieve items from multiple tables
+   * create an instance of BatchGetItemInput and use store.makeRequest with it.
    */
   batchGetItem(keys: any[]): BatchGetSingleTableRequest<T> {
     return new BatchGetSingleTableRequest(this.dynamoRx, this.modelClazz, this.tableName, keys)
diff --git a/src/dynamo/request/batchwritesingletable/batch-write-single-table.request.spec.ts b/src/dynamo/request/batchwritesingletable/batch-write-single-table.request.spec.ts
@@ -0,0 +1,126 @@
+import * as DynamoDB from 'aws-sdk/clients/dynamodb'
+import * as moment from 'moment'
+import { of } from 'rxjs'
+import { getTableName } from '../../../../test/helper/get-table-name.function'
+import { Organization } from '../../../../test/models/organization.model'
+import { DEFAULT_SESSION_VALIDITY_ENSURER } from '../../default-session-validity-ensurer.const'
+import { DynamoRx } from '../../dynamo-rx'
+import { BatchWriteSingleTableRequest } from './batch-write-single-table.request'
+
+describe('batch write single table request', () => {
+  const tableName = getTableName(Organization)
+
+  let item: Organization
+  let dynamoRx: DynamoRx
+  let request: BatchWriteSingleTableRequest<Organization>
+
+  let nextSpyFn: () => { value: number }
+  const generatorMock = () => <any>{ next: nextSpyFn }
+
+  beforeEach(() => {
+    item = <any>{
+      id: 'myId',
+      createdAtDate: moment(),
+      name: 'myOrg',
+    }
+    nextSpyFn = jest.fn().mockImplementation(() => ({ value: 0 }))
+  })
+
+  describe('correct params', () => {
+    beforeEach(() => {
+      dynamoRx = new DynamoRx(DEFAULT_SESSION_VALIDITY_ENSURER)
+      request = new BatchWriteSingleTableRequest(dynamoRx, Organization, tableName)
+
+      const output: DynamoDB.BatchWriteItemOutput = {}
+      spyOn(dynamoRx, 'batchWriteItem').and.returnValue(of(output))
+    })
+
+    it('delete with complex primary key', async () => {
+      request.delete([item])
+      await request.exec(generatorMock).toPromise()
+
+      expect(dynamoRx.batchWriteItem).toHaveBeenCalledTimes(1)
+      expect(dynamoRx.batchWriteItem).toHaveBeenCalledWith({
+        RequestItems: {
+          [tableName]: [
+            {
+              DeleteRequest: {
+                Key: {
+                  id: { S: 'myId' },
+                  createdAtDate: { S: item.createdAtDate.utc().format() },
+                },
+              },
+            },
+          ],
+        },
+      })
+      expect(nextSpyFn).toHaveBeenCalledTimes(0)
+    })
+
+    it('put object', async () => {
+      request.put([item])
+      await request.exec(generatorMock).toPromise()
+
+      expect(dynamoRx.batchWriteItem).toHaveBeenCalledTimes(1)
+      expect(dynamoRx.batchWriteItem).toHaveBeenCalledWith({
+        RequestItems: {
+          [tableName]: [
+            {
+              PutRequest: {
+                Item: {
+                  id: { S: 'myId' },
+                  createdAtDate: { S: item.createdAtDate.utc().format() },
+                  name: { S: 'myOrg' },
+                },
+              },
+            },
+          ],
+        },
+      })
+      expect(nextSpyFn).toHaveBeenCalledTimes(0)
+    })
+
+    it('delete >25 items in two requests', async () => {
+      const twentyFiveItems = []
+      for (let i = 0; i < 25; i++) {
+        twentyFiveItems.push(item)
+      }
+      request.delete(twentyFiveItems)
+      request.delete(twentyFiveItems)
+      await request.exec(generatorMock).toPromise()
+      expect(dynamoRx.batchWriteItem).toHaveBeenCalledTimes(2)
+      expect(nextSpyFn).toHaveBeenCalledTimes(0)
+    })
+  })
+
+  describe('correct backoff', () => {
+    beforeEach(() => {
+      dynamoRx = new DynamoRx(DEFAULT_SESSION_VALIDITY_ENSURER)
+      request = new BatchWriteSingleTableRequest(dynamoRx, Organization, tableName)
+
+      const output: DynamoDB.BatchWriteItemOutput = {
+        UnprocessedItems: {
+          [tableName]: [
+            {
+              PutRequest: {
+                Item: {
+                  id: { S: 'myId' },
+                  createdAtDate: { S: item.createdAtDate.utc().format() },
+                  name: { S: 'myOrg' },
+                },
+              },
+            },
+          ],
+        },
+      }
+      spyOn(dynamoRx, 'batchWriteItem').and.returnValues(of(output), of({}))
+    })
+
+    it('should retry when capacity is exceeded', async () => {
+      request.put([item])
+      await request.exec(generatorMock).toPromise()
+      expect(dynamoRx.batchWriteItem).toHaveBeenCalledTimes(2)
+      expect(nextSpyFn).toHaveBeenCalledTimes(1)
+    })
+  })
+})
diff --git a/src/dynamo/request/batchwritesingletable/batch-write-single-table.request.ts b/src/dynamo/request/batchwritesingletable/batch-write-single-table.request.ts
@@ -0,0 +1,106 @@
+import {
+  AttributeMap,
+  BatchWriteItemInput,
+  BatchWriteItemOutput,
+  WriteRequest,
+  WriteRequests,
+} from 'aws-sdk/clients/dynamodb'
+import { Observable, of } from 'rxjs'
+import { delay, map, mergeMap, tap } from 'rxjs/operators'
+import { DynamoRx } from '../../../dynamo/dynamo-rx'
+import { randomExponentialBackoffTimer } from '../../../helper'
+import { Mapper } from '../../../mapper'
+import { ModelConstructor } from '../../../model/model-constructor'
+import { BatchWriteSingleTableResponse } from './batch-write-single-table.response'
+
+const MAX_BATCH_WRITE_ITEMS = 25
+
+export class BatchWriteSingleTableRequest<T> {
+  private get toKey(): (item: T) => AttributeMap {
+    if (!this._keyFn) {
+      this._keyFn = Mapper.createToKeyFn(this.modelClazz)
+    }
+    return this._keyFn
+  }
+
+  readonly dynamoRx: DynamoRx
+  readonly modelClazz: ModelConstructor<T>
+  readonly tableName: string
+  readonly itemsToProcess: WriteRequests
+
+  private _keyFn: any
+
+  constructor(dynamoRx: DynamoRx, modelClazz: ModelConstructor<T>, tableName: string) {
+    this.dynamoRx = dynamoRx
+
+    if (modelClazz === null || modelClazz === undefined) {
+      throw new Error("please provide the model clazz for the request, won't work otherwise")
+    }
+    this.modelClazz = modelClazz
+    this.tableName = tableName
+
+    this.itemsToProcess = []
+  }
+
+  delete(items: T[]): BatchWriteSingleTableRequest<T> {
+    this.itemsToProcess.push(...items.map<WriteRequest>(item => ({ DeleteRequest: { Key: this.toKey(item) } })))
+    return this
+  }
+
+  put(items: T[]): BatchWriteSingleTableRequest<T> {
+    this.itemsToProcess.push(
+      ...items.map<WriteRequest>(item => ({ PutRequest: { Item: Mapper.toDb(item, this.modelClazz) } }))
+    )
+    return this
+  }
+
+  private execNextBatch(): Observable<BatchWriteSingleTableResponse> {
+    const batch = this.itemsToProcess.splice(0, MAX_BATCH_WRITE_ITEMS)
+    const batchWriteItemInput: BatchWriteItemInput = {
+      RequestItems: {
+        [this.tableName]: batch,
+      },
+    }
+
+    return this.dynamoRx.batchWriteItem(batchWriteItemInput).pipe(
+      tap((batchWriteManyResponse: BatchWriteItemOutput) => {
+        if (batchWriteManyResponse.UnprocessedItems) {
+          this.itemsToProcess.unshift(...batchWriteManyResponse.UnprocessedItems[this.tableName])
+        }
+      }),
+      map((batchWriteManyResponse: BatchWriteItemOutput) => ({
+        remainingItems: this.itemsToProcess.length,
+        capacityExceeded: !!batchWriteManyResponse.UnprocessedItems,
+        consumedCapacity: batchWriteManyResponse.ConsumedCapacity,
+      }))
+    )
+  }
+
+  /**
+   *
+   * @param backoffTimer generator for how much timeSlots should be waited before requesting next batch. only used when capacity was exceeded. default randomExponentialBackoffTimer
+   * @param throttleTimeSlot defines how long one timeSlot is for throttling, default 1 second
+   */
+  exec(backoffTimer = randomExponentialBackoffTimer, throttleTimeSlot = 1000): Observable<void> {
+    let rBoT = backoffTimer()
+    let backoffTime = 0
+    return this.execNextBatch().pipe(
+      mergeMap((r: BatchWriteSingleTableResponse) => {
+        if (!r.capacityExceeded) {
+          rBoT = backoffTimer()
+          backoffTime = 0
+        } else {
+          backoffTime = rBoT.next().value
+        }
+        return of(r).pipe(delay(backoffTime * throttleTimeSlot))
+      }),
+      mergeMap((r: BatchWriteSingleTableResponse) => {
+        if (r.remainingItems > 0) {
+          return this.exec()
+        } else {
+          return of()
+        }
+      })
+    )
+  }
+}
diff --git a/src/dynamo/request/batchwritesingletable/batch-write-single-table.response.ts b/src/dynamo/request/batchwritesingletable/batch-write-single-table.response.ts
@@ -0,0 +1,7 @@
+import { ConsumedCapacityMultiple } from 'aws-sdk/clients/dynamodb'
+
+export interface BatchWriteSingleTableResponse {
+  remainingItems: number
+  capacityExceeded: boolean
+  consumedCapacity?: ConsumedCapacityMultiple
+}
diff --git a/src/dynamo/request/query/query.request.ts b/src/dynamo/request/query/query.request.ts
@@ -1,6 +1,7 @@
 import { QueryInput, QueryOutput } from 'aws-sdk/clients/dynamodb'
 import { Observable } from 'rxjs'
 import { map } from 'rxjs/operators'
+import { fetchAll } from '../../../helper'
 import { Mapper } from '../../../mapper/mapper'
 import { ModelConstructor } from '../../../model/model-constructor'
 import { DynamoRx } from '../../dynamo-rx'
@@ -132,4 +133,11 @@ export class QueryRequest<T> extends Request<T, QueryRequest<T>, QueryInput, Que
       })
     )
   }
+
+  /**
+   * fetches all pages. may uses all provisionedOutput, therefore for client side use cases rather use pagedDatasource (exec)
+   */
+  execFetchAll(): Observable<T[]> {
+    return fetchAll(this)
+  }
 }
diff --git a/src/dynamo/request/scan/scan.request.spec.ts b/src/dynamo/request/scan/scan.request.spec.ts
@@ -1,13 +1,13 @@
 import { ScanInput, ScanOutput } from 'aws-sdk/clients/dynamodb'
-import { Observable } from 'rxjs'
+import { Observable, of } from 'rxjs'
 import { getTableName } from '../../../../test/helper/get-table-name.function'
 import { ComplexModel } from '../../../../test/models/complex.model'
 import { Request } from '../request.model'
 import { ScanRequest } from './scan.request'
 
 class DynamoRxMock {
   scan(): Observable<ScanOutput> {
-    return Observable.of({ Count: 10 })
+    return of({ Count: 10 })
   }
 }
 
diff --git a/src/dynamo/request/scan/scan.request.ts b/src/dynamo/request/scan/scan.request.ts
@@ -1,6 +1,7 @@
 import { ScanInput, ScanOutput } from 'aws-sdk/clients/dynamodb'
 import { Observable } from 'rxjs'
 import { map } from 'rxjs/operators'
+import { fetchAll } from '../../../helper'
 import { Mapper } from '../../../mapper/mapper'
 import { ModelConstructor } from '../../../model/model-constructor'
 import { DynamoRx } from '../../dynamo-rx'
@@ -66,4 +67,11 @@ export class ScanRequest<T> extends Request<T, ScanRequest<T>, ScanInput, ScanRe
 
     return this.dynamoRx.scan(params).pipe(map(response => response.Count!))
   }
+
+  /**
+   * fetches all pages. may uses all provisionedOutput, therefore for client side use cases rather use pagedDatasource (exec)
+   */
+  execFetchAll(): Observable<T[]> {
+    return fetchAll(this)
+  }
 }
diff --git a/src/helper/fetch-all.function.ts b/src/helper/fetch-all.function.ts
@@ -0,0 +1,28 @@
+import { Key } from 'aws-sdk/clients/dynamodb'
+import { Observable, of } from 'rxjs'
+import { concatMap, map } from 'rxjs/operators'
+import { QueryRequest, Request, ScanRequest, ScanResponse } from '../dynamo/request'
+
+/**
+ * When we cant load all the items of a table with one request, we will fetch as long as there is more data
+ * available. This can be used with scan and query requests.
+ */
+
+export function fetchAll<T>(request: ScanRequest<T> | QueryRequest<T>, startKey?: Key): Observable<T[]> {
+  request.limit(Request.INFINITE_LIMIT)
+  if (startKey) {
+    request.exclusiveStartKey(startKey)
+  }
+  return request.execFullResponse().pipe(
+    // tap(response => console.debug('response', response)),
+    concatMap<ScanResponse<T>, T[]>(response => {
+      if (response.LastEvaluatedKey) {
+        return fetchAll(request, response.LastEvaluatedKey).pipe(
+          map(innerResponse => response.Items!.concat(innerResponse))
+        )
+      } else {
+        return of(response.Items)
+      }
+    })
+  )
+}
diff --git a/src/helper/index.ts b/src/helper/index.ts
@@ -0,0 +1,2 @@
+export * from './fetch-all.function'
+export * from './random-exponential-backoff-timer.generator'
diff --git a/src/helper/random-exponential-backoff-timer.generator.ts b/src/helper/random-exponential-backoff-timer.generator.ts
@@ -0,0 +1,6 @@
+export function* randomExponentialBackoffTimer() {
+  let i = 0
+  while (true) {
+    yield (Math.pow(2, Math.round(Math.random() * ++i)) - 1) / 2
+  }
+}
diff --git a/src/mapper/mapper.ts b/src/mapper/mapper.ts
diff --git a/src/model/model-constructor.ts b/src/model/model-constructor.ts
diff --git a/tslint.json b/tslint.json

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,7 @@`
`1`	`1`	`import { QueryInput, QueryOutput } from 'aws-sdk/clients/dynamodb'`
`2`	`2`	`import { Observable } from 'rxjs'`
`3`	`3`	`import { map } from 'rxjs/operators'`
	`4`	`+import { fetchAll } from '../../../helper'`
`4`	`5`	`import { Mapper } from '../../../mapper/mapper'`
`5`	`6`	`import { ModelConstructor } from '../../../model/model-constructor'`
`6`	`7`	`import { DynamoRx } from '../../dynamo-rx'`
`@@ -132,4 +133,11 @@ export class QueryRequest<T> extends Request<T, QueryRequest<T>, QueryInput, Que`
`132`	`133`	`})`
`133`	`134`	`)`
`134`	`135`	`}`
	`136`	`+`
	`137`	`+ /**`
	`138`	`+ * fetches all pages. may uses all provisionedOutput, therefore for client side use cases rather use pagedDatasource (exec)`
	`139`	`+ */`
	`140`	`+ execFetchAll(): Observable<T[]> {`
	`141`	`+ return fetchAll(this)`
	`142`	`+ }`
`135`	`143`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,13 +1,13 @@`
`1`	`1`	`import { ScanInput, ScanOutput } from 'aws-sdk/clients/dynamodb'`
`2`		`-import { Observable } from 'rxjs'`
	`2`	`+import { Observable, of } from 'rxjs'`
`3`	`3`	`import { getTableName } from '../../../../test/helper/get-table-name.function'`
`4`	`4`	`import { ComplexModel } from '../../../../test/models/complex.model'`
`5`	`5`	`import { Request } from '../request.model'`
`6`	`6`	`import { ScanRequest } from './scan.request'`
`7`	`7`
`8`	`8`	`class DynamoRxMock {`
`9`	`9`	`scan(): Observable<ScanOutput> {`
`10`		`- return Observable.of({ Count: 10 })`
	`10`	`+ return of({ Count: 10 })`
`11`	`11`	`}`
`12`	`12`	`}`
`13`	`13`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+export * from './fetch-all.function'`
	`2`	`+export * from './random-exponential-backoff-timer.generator'`