Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 | 15x 106x 106x 8x 8x 8x 8x 4x 4x 4x 4x 4x 8x 8x 4x 4x 4x 4x 4x 4x 8x 8x 8x | import { AppSyncClient } from '../common_services';
import {
DeviceStatusCheckFn,
DbStreamConverterFn,
DeviceStatus,
} from '../types.d';
export abstract class UpdateEventPublisher {
constructor(
private readonly appSyncClient$: AppSyncClient,
public readonly checkDeviceStatus: DeviceStatusCheckFn,
) {}
abstract prepareUpdateEventRequest(item: any, keys: any): any;
/**
*
* @param stream stream of update events coming from DynamoDb
* @param streamConverterFn Function used to convert the event records to type interfaces
*/
async publishUpdateEventToSubscribers<T>(
stream: Array<any>,
streamConverterFn: DbStreamConverterFn<T>,
) {
const client = await this.appSyncClient$.getAppSyncClient();
const records: any[] = [];
const deviceStatus: Promise<any>[] = stream
.filter((record: any) => typeof record.dynamodb.NewImage !== 'undefined')
.map((record: any) => {
const { dynamodb } = record;
const { Keys: keys } = dynamodb;
const deviceUuid = keys.id.S;
records.push(record);
return this.checkDeviceStatus(deviceUuid);
});
const status = await Promise.all(deviceStatus);
const updatedRecords = records
.filter((_, i: number) => status[i] === DeviceStatus.ACTIVE)
.map((record: any): T => streamConverterFn(record))
.map((item: T) => {
console.info('publish update event', item);
const keys = this.appSyncClient$.utils.getGqlResponseKeys(item);
console.info('publish keys:', keys);
return client.mutate(this.prepareUpdateEventRequest(item, keys));
});
try {
const response = await Promise.all(updatedRecords);
console.info('publish event response,', JSON.stringify(response));
} catch (err) {
// TODO: send notification for this error
console.error(err);
}
}
}
|