smartuniverse/ts/smartuniverse.classes.universecache.ts

81 lines
2.3 KiB
TypeScript
Raw Permalink Normal View History

2018-03-13 05:15:40 +00:00
import * as plugins from './smartuniverse.plugins';
import { UniverseChannel } from './smartuniverse.classes.universechannel';
2018-03-13 05:15:40 +00:00
import { UniverseMessage } from './smartuniverse.classes.universemessage';
2019-01-31 01:52:18 +00:00
import { Objectmap } from '@pushrocks/lik';
2018-03-13 05:15:40 +00:00
2019-01-31 01:52:18 +00:00
import { Observable, from } from 'rxjs';
import { filter } from 'rxjs/operators';
import { rxjs } from '@pushrocks/smartrx';
2019-08-12 12:59:37 +00:00
import { UniverseConnection } from './smartuniverse.classes.universeconnection';
2019-08-13 13:48:20 +00:00
import { Universe } from './smartuniverse.classes.universe';
2018-03-13 05:15:40 +00:00
2018-04-29 14:38:23 +00:00
/**
* universe store handles the creation, storage and retrieval of messages.
*/
export class UniverseCache {
// ========
// INSTANCE
// ========
2018-03-13 05:15:40 +00:00
public standardMessageExpiry: number;
2019-09-17 13:40:54 +00:00
public destructionTime: number = 10000;
/**
* stores messages for this instance
*/
2019-09-10 21:55:20 +00:00
public messageMap = new Objectmap<UniverseMessage<any>>();
/**
* stores the channels that are available within the universe
*/
public channelMap = new Objectmap<UniverseChannel>();
2019-08-13 13:48:20 +00:00
/**
* stores all connections
*/
2019-08-12 12:59:37 +00:00
public connectionMap = new plugins.lik.Objectmap<UniverseConnection>();
/**
* allows messages to be processed in a blacklist mode for further analysis
*/
2019-08-13 13:48:20 +00:00
public blackListChannel: UniverseChannel;
public universeRef: Universe;
2019-08-13 13:48:20 +00:00
constructor(universeArg: Universe, standardMessageExpiryArg: number) {
this.universeRef = universeArg;
2018-03-13 05:15:40 +00:00
this.standardMessageExpiry = standardMessageExpiryArg;
2019-08-13 13:48:20 +00:00
this.blackListChannel = new UniverseChannel(this.universeRef, 'blacklist', 'nada');
2018-03-13 05:15:40 +00:00
}
/**
2019-04-11 15:52:01 +00:00
* add a message to the UniverseCache
2018-03-13 05:15:40 +00:00
* @param messageArg
* @param attachedPayloadArg
*/
2019-09-10 21:55:20 +00:00
public async addMessage(messageArg: UniverseMessage<any>) {
messageArg.setUniverseCache(this);
UniverseChannel.authorizeAMessageForAChannel(this, messageArg);
this.messageMap.add(messageArg);
2019-08-13 13:48:20 +00:00
messageArg.universeChannelList.forEach(universeChannel => {
2019-09-10 07:56:32 +00:00
universeChannel.push(messageArg);
2019-08-13 13:48:20 +00:00
});
2018-03-13 05:15:40 +00:00
}
/**
2019-04-11 15:52:01 +00:00
* Read a message from the UniverseCache
2018-03-13 05:15:40 +00:00
*/
2019-06-06 21:23:37 +00:00
public readMessagesYoungerThan(
unixTimeArg?: number,
channelName?: string
2019-09-10 21:55:20 +00:00
): Observable<UniverseMessage<any>> {
2019-01-31 01:52:18 +00:00
const messageObservable = from(this.messageMap.getArray()).pipe(
filter(messageArg => {
2019-04-22 07:58:36 +00:00
return messageArg.smartTimestamp.isYoungerThanMilliSeconds(this.destructionTime);
2019-01-31 01:52:18 +00:00
})
2018-03-20 07:16:54 +00:00
);
2018-03-13 05:15:40 +00:00
return messageObservable;
}
}