Merge branch 'beta' into #16-updateExpression-ifnotexists

Author: michaelwittwer

dynamoDbAbstraction.monopic

@@ -0,0 +1,2 @@
+export const BATCH_GET_MAX_REQUEST_ITEM_COUNT = 100
+export const BATCH_GET_DEFAULT_TIME_SLOT = 1000

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

@@ -71,8 +71,8 @@ describe('batch get', () => {
     })
 
     it('should throw when more than 100 items are added', () => {
-      const items55: Array<Partial<SimpleWithPartitionKeyModel>> = new Array(55).map((x, i) => ({ id: `id-${i}` }))
-      const items60: Array<Partial<Organization>> = new Array(60).map((x, i) => ({
+      const items55: Array<Partial<SimpleWithPartitionKeyModel>> = [...new Array(55)].map((x, i) => ({ id: `id-${i}` }))
+      const items60: Array<Partial<Organization>> = [...new Array(60)].map((x, i) => ({
         id: `id-${i}`,
         createdAtDate: new Date(),
       }))

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

@@ -10,10 +10,10 @@ import { DynamoRx } from '../dynamo-rx'
 import { getTableName } from '../get-table-name.function'
 import { BatchGetFullResponse } from './batch-get-full.response'
 import { batchGetItemsFetchAll } from './batch-get-utils'
+import { BATCH_GET_DEFAULT_TIME_SLOT, BATCH_GET_MAX_REQUEST_ITEM_COUNT } from './batch-get.const'
 import { BatchGetResponse } from './batch-get.response'
 
-const MAX_REQUEST_ITEM_COUNT = 100
-const DEFAULT_TIME_SLOT = 1000
+
 
 export class BatchGetRequest {
   readonly params: DynamoDB.BatchGetItemInput
@@ -46,7 +46,7 @@ export class BatchGetRequest {
     this.tables.set(tableName, modelClazz)
 
     // check if keys to add do not exceed max count
-    if (this.itemCounter + keys.length > MAX_REQUEST_ITEM_COUNT) { throw new Error(`you can request at max ${MAX_REQUEST_ITEM_COUNT} items per request`)}
+    if (this.itemCounter + keys.length > BATCH_GET_MAX_REQUEST_ITEM_COUNT) { throw new Error(`you can request at max ${BATCH_GET_MAX_REQUEST_ITEM_COUNT} items per request`)}
 
     this.params.RequestItems[tableName] = {
       Keys: keys.map(createToKeyFn(modelClazz)),
@@ -58,26 +58,26 @@ export class BatchGetRequest {
     return this
   }
 
-  execNoMap(backoffTimer = randomExponentialBackoffTimer, throttleTimeSlot = DEFAULT_TIME_SLOT): Observable<DynamoDB.BatchGetItemOutput> {
+  execNoMap(backoffTimer = randomExponentialBackoffTimer, throttleTimeSlot = BATCH_GET_DEFAULT_TIME_SLOT): Observable<DynamoDB.BatchGetItemOutput> {
     return this.fetch(backoffTimer, throttleTimeSlot)
   }
 
-  execFullResponse(backoffTimer = randomExponentialBackoffTimer, throttleTimeSlot = DEFAULT_TIME_SLOT): Observable<BatchGetFullResponse> {
+  execFullResponse(backoffTimer = randomExponentialBackoffTimer, throttleTimeSlot = BATCH_GET_DEFAULT_TIME_SLOT): Observable<BatchGetFullResponse> {
     return this.fetch(backoffTimer, throttleTimeSlot)
       .pipe(
         map(this.mapResponse),
       )
   }
 
-  exec(backoffTimer = randomExponentialBackoffTimer, throttleTimeSlot = DEFAULT_TIME_SLOT): Observable<BatchGetResponse> {
+  exec(backoffTimer = randomExponentialBackoffTimer, throttleTimeSlot = BATCH_GET_DEFAULT_TIME_SLOT): Observable<BatchGetResponse> {
     return this.fetch(backoffTimer, throttleTimeSlot)
       .pipe(
         map(this.mapResponse),
         map(r => r.Responses),
       )
   }
 
-  private fetch(backoffTimer = randomExponentialBackoffTimer, throttleTimeSlot = DEFAULT_TIME_SLOT) {
+  private fetch(backoffTimer = randomExponentialBackoffTimer, throttleTimeSlot = BATCH_GET_DEFAULT_TIME_SLOT) {
     return batchGetItemsFetchAll(this.dynamoRx, { ...this.params }, backoffTimer(), throttleTimeSlot)
   }
 

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

@@ -0,0 +1,54 @@
+import { DynamoDB } from 'aws-sdk'
+import { of } from 'rxjs'
+import { DynamoRx } from '../dynamo-rx'
+import { batchWriteItemsWriteAll, hasUnprocessedItems } from './batch-write-utils'
+
+describe('batch-write-utils', () => {
+  describe('batchWriteItemsWriteAll', () => {
+    let batchWriteItemSpy: jasmine.Spy
+    let dynamoRx: DynamoRx
+    let backoffTimerMock: { next: jasmine.Spy }
+
+    const output1: DynamoDB.BatchWriteItemOutput = {
+      UnprocessedItems: {
+        tableA: [
+          {
+            PutRequest: { Item: { id: { S: 'id-A' } } },
+          },
+        ],
+      },
+    }
+    const output2: DynamoDB.BatchWriteItemOutput = {}
+
+    beforeEach(async () => {
+      batchWriteItemSpy = jasmine.createSpy().and.returnValues(of(output1), of(output2))
+      dynamoRx = <any>{ batchWriteItem: batchWriteItemSpy }
+      backoffTimerMock = { next: jasmine.createSpy().and.returnValue({ value: 0 }) }
+
+      await batchWriteItemsWriteAll(dynamoRx, <any>{}, <IterableIterator<number>>(<any>backoffTimerMock), 0).toPromise()
+    })
+
+    it('should use UnprocessedKeys for next request', () => {
+      expect(batchWriteItemSpy).toHaveBeenCalledTimes(2)
+      expect(batchWriteItemSpy.calls.mostRecent().args[0]).toBeDefined()
+      expect(batchWriteItemSpy.calls.mostRecent().args[0].RequestItems).toBeDefined()
+      expect(batchWriteItemSpy.calls.mostRecent().args[0].RequestItems).toEqual(output1.UnprocessedItems)
+    })
+
+    it('should backoff when UnprocessedItems', () => {
+      expect(backoffTimerMock.next).toHaveBeenCalledTimes(1)
+    })
+  })
+
+  describe('hasUnprocessedItems', () => {
+    it('should return false when no items', () => {
+      expect(hasUnprocessedItems({})).toBe(false)
+      expect(hasUnprocessedItems({ UnprocessedItems: {} })).toBe(false)
+      expect(hasUnprocessedItems({ UnprocessedItems: { myTable: [] } })).toBe(false)
+    })
+
+    it('should return true when items', () => {
+      expect(hasUnprocessedItems({ UnprocessedItems: { myTable: [{ PutRequest: { Item: { id: { S: 'myId' } } } }] } }))
+    })
+  })
+})

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

@@ -1,7 +1,7 @@
 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 { DynamoRx } from '../dynamo-rx'
 
 
 export function batchWriteItemsWriteAll(

src/dynamo/request/batchwritesingletable/batch-write-utils.ts renamed to src/dynamo/batchwrite/batch-write-utils.ts

@@ -0,0 +1,2 @@
+export const BATCH_WRITE_MAX_REQUEST_ITEM_COUNT = 25
+export const BATCH_WRITE_DEFAULT_TIME_SLOT = 1000

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

@@ -0,0 +1,150 @@
+import { of } from 'rxjs'
+import { ComplexModel, SimpleWithPartitionKeyModel } from '../../../test/models'
+import { getTableName } from '../get-table-name.function'
+import { BatchWriteRequest } from './batch-write.request'
+
+describe('batchWriteRequest', () => {
+  let req: BatchWriteRequest
+
+  describe('constructor', () => {
+    it('should initialize params', () => {
+      req = new BatchWriteRequest()
+      expect(req.params.RequestItems).toBeDefined()
+      expect(req.params.RequestItems).toEqual({})
+    })
+  })
+
+  describe('returnConsumedCapacity', () => {
+    it('should set params', () => {
+      req = new BatchWriteRequest().returnConsumedCapacity('TOTAL')
+      expect(req.params.ReturnConsumedCapacity).toBe('TOTAL')
+    })
+  })
+
+  describe('returnItemCollectionMetrics', () => {
+    it('should set params', () => {
+      req = new BatchWriteRequest().returnItemCollectionMetrics('SIZE')
+      expect(req.params.ReturnItemCollectionMetrics).toBe('SIZE')
+    })
+  })
+
+  describe('delete', () => {
+    const now = new Date()
+
+    it('should set params', () => {
+      req = new BatchWriteRequest()
+        .delete(SimpleWithPartitionKeyModel, [{ id: 'myId1' }])
+        .delete(ComplexModel, [{ id: 'myId2', creationDate: now }])
+
+      expect(Object.keys(req.params.RequestItems)).toEqual([
+        getTableName(SimpleWithPartitionKeyModel),
+        getTableName(ComplexModel),
+      ])
+
+      const simpleReqItems = req.params.RequestItems[getTableName(SimpleWithPartitionKeyModel)]
+      expect(simpleReqItems).toBeDefined()
+      expect(simpleReqItems.length).toBe(1)
+      expect(simpleReqItems[0].DeleteRequest).toEqual({
+        Key: { id: { S: 'myId1' } },
+      })
+
+      const complexReqItems = req.params.RequestItems[getTableName(ComplexModel)]
+      expect(complexReqItems).toBeDefined()
+      expect(complexReqItems.length).toBe(1)
+      expect(complexReqItems[0].DeleteRequest).toEqual({
+        Key: { id: { S: 'myId2' }, creationDate: { S: now.toISOString() } },
+      })
+    })
+
+    it('should throw when too many items', () => {
+      const items: Array<Partial<SimpleWithPartitionKeyModel>> = [...new Array(30)].map((_, ix) => ({
+        id: `mydId-${ix}`,
+      }))
+      expect(() => new BatchWriteRequest().delete(SimpleWithPartitionKeyModel, items)).toThrow()
+    })
+  })
+
+  describe('put', () => {
+    const now = new Date()
+
+    it('should set params', () => {
+      req = new BatchWriteRequest()
+        .put(SimpleWithPartitionKeyModel, [{ id: 'myId1' }])
+        .put(ComplexModel, [{ id: 'myId2', creationDate: now }])
+
+      const simpleReqItems = req.params.RequestItems[getTableName(SimpleWithPartitionKeyModel)]
+      expect(simpleReqItems).toBeDefined()
+      expect(simpleReqItems.length).toBe(1)
+      expect(simpleReqItems[0].PutRequest).toEqual({
+        Item: { id: { S: 'myId1' } },
+      })
+
+      const complexReqItems = req.params.RequestItems[getTableName(ComplexModel)]
+      expect(complexReqItems).toBeDefined()
+      expect(complexReqItems.length).toBe(1)
+      expect(complexReqItems[0].PutRequest).toEqual({
+        Item: { id: { S: 'myId2' }, creationDate: { S: now.toISOString() } },
+      })
+    })
+
+    it('should throw when too many items', () => {
+      const items: SimpleWithPartitionKeyModel[] = [...new Array(30)].map((_, ix) => ({ id: `mydId-${ix}`, age: ix }))
+      expect(() => new BatchWriteRequest().put(SimpleWithPartitionKeyModel, items)).toThrow()
+    })
+  })
+
+  describe('put and delete mixed', () => {
+    const now = new Date()
+    const tnSimple = getTableName(SimpleWithPartitionKeyModel)
+    const tnComplex = getTableName(ComplexModel)
+
+    const simpleItem: SimpleWithPartitionKeyModel = { id: 'myId', age: 25 }
+    const simpleItems: SimpleWithPartitionKeyModel[] = [...new Array(16)].map((_, ix) => ({
+      id: `mydId-${ix}`,
+      age: ix,
+    }))
+    const complexItems: ComplexModel[] = [...new Array(16)].map(
+      (_, ix) => <ComplexModel>{ id: `myId-${ix}`, creationDate: now },
+    )
+
+    beforeEach(() => (req = new BatchWriteRequest()))
+
+    it('should add correct request items', () => {
+      req.put(ComplexModel, complexItems).delete(SimpleWithPartitionKeyModel, [simpleItem])
+      expect(req.params.RequestItems[tnComplex].length).toBe(16)
+      expect(req.params.RequestItems[tnComplex][5].PutRequest).toBeDefined()
+
+      expect(req.params.RequestItems[tnSimple].length).toBe(1)
+      expect(req.params.RequestItems[tnSimple][0].DeleteRequest).toBeDefined()
+    })
+
+    it('should throw when too many (1)', () => {
+      expect(() => req.put(ComplexModel, complexItems).delete(SimpleWithPartitionKeyModel, simpleItems)).toThrow()
+    })
+    it('should throw when too many (2)', () => {
+      expect(() => req.delete(ComplexModel, complexItems).put(SimpleWithPartitionKeyModel, simpleItems)).toThrow()
+    })
+  })
+
+  describe('exec functions', () => {
+    let batchWriteItemSpy: jasmine.Spy
+
+    beforeEach(() => {
+      const output = {
+        myResponse: true,
+      }
+      batchWriteItemSpy = jasmine.createSpy().and.returnValue(of(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()
+    })
+
+    it('execFullResponse should return the full response', async () => {
+      expect(await req.execFullResponse().toPromise()).toEqual({ myResponse: true })
+    })
+  })
+})

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

@@ -0,0 +1,83 @@
+import { DynamoDB } from 'aws-sdk'
+import { Observable } from 'rxjs'
+import { map } from 'rxjs/operators'
+import { randomExponentialBackoffTimer } from '../../helper'
+import { createToKeyFn, toDb } from '../../mapper'
+import { ModelConstructor } from '../../model'
+import { DynamoRx } from '../dynamo-rx'
+import { getTableName } from '../get-table-name.function'
+import { batchWriteItemsWriteAll } from './batch-write-utils'
+import { BATCH_WRITE_DEFAULT_TIME_SLOT, BATCH_WRITE_MAX_REQUEST_ITEM_COUNT } from './batch-write.const'
+
+export class BatchWriteRequest {
+  readonly params: DynamoDB.BatchWriteItemInput
+  private readonly dynamoRx: DynamoRx
+  private itemCount = 0
+
+  constructor() {
+    this.dynamoRx = new DynamoRx()
+    this.params = {
+      RequestItems: {},
+    }
+  }
+
+  returnConsumedCapacity(value: DynamoDB.ReturnConsumedCapacity): BatchWriteRequest {
+    this.params.ReturnConsumedCapacity = value
+    return this
+  }
+
+  returnItemCollectionMetrics(value: DynamoDB.ReturnItemCollectionMetrics): BatchWriteRequest {
+    this.params.ReturnItemCollectionMetrics = value
+    return this
+  }
+
+  delete<T>(modelClazz: ModelConstructor<T>, items: Array<Partial<T>>): BatchWriteRequest {
+    this.requestItems(modelClazz, items.map(this.createDeleteRequest(modelClazz)))
+    return this
+  }
+
+  put<T>(modelClazz: ModelConstructor<T>, items: T[]): BatchWriteRequest {
+    this.requestItems(modelClazz, items.map(this.createPutRequest(modelClazz)))
+    return this
+  }
+
+  /**
+   *
+   * @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 = BATCH_WRITE_DEFAULT_TIME_SLOT): Observable<void> {
+    return this.write(backoffTimer, throttleTimeSlot).pipe(
+      map(() => {
+        return
+      }),
+    )
+  }
+
+  execFullResponse(backoffTimer = randomExponentialBackoffTimer, throttleTimeSlot = BATCH_WRITE_DEFAULT_TIME_SLOT): Observable<DynamoDB.BatchWriteItemOutput> {
+    return this.write(backoffTimer, throttleTimeSlot)
+  }
+
+  private requestItems(modelClazz: ModelConstructor<any>, items: DynamoDB.WriteRequests) {
+    if (this.itemCount + items.length > BATCH_WRITE_MAX_REQUEST_ITEM_COUNT) {
+      throw new Error(`batch write takes at max ${BATCH_WRITE_MAX_REQUEST_ITEM_COUNT} items`)
+    }
+    const tableName = getTableName(modelClazz)
+    this.params.RequestItems[tableName] = this.params.RequestItems[tableName] || []
+    this.params.RequestItems[tableName].push(...items)
+    this.itemCount += items.length
+  }
+
+  private write(backoffTimer: () => IterableIterator<number>, throttleTimeSlot: number) {
+    return batchWriteItemsWriteAll(this.dynamoRx, { ...this.params }, backoffTimer(), throttleTimeSlot)
+  }
+
+  private createDeleteRequest = <T>(modelClazz: ModelConstructor<T>) => {
+    const toKey = createToKeyFn(modelClazz)
+    return (item: Partial<T>): DynamoDB.WriteRequest => ({ DeleteRequest: { Key: toKey(item) } })
+  }
+
+  private createPutRequest = <T>(modelClazz: ModelConstructor<T>) => {
+    return (item: T): DynamoDB.WriteRequest => ({ PutRequest: { Item: toDb(item, modelClazz) } })
+  }
+}

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

@@ -0,0 +1 @@
+export * from './batch-write.request'