|
|
|
@ -33,9 +33,9 @@ import { |
|
|
|
TsValue |
|
|
|
} from '@shared/models/query/query.models'; |
|
|
|
import { PageData } from '@shared/models/page/page-data'; |
|
|
|
import { alarmFields } from '@shared/models/alarm.models'; |
|
|
|
import { alarmFields, AlarmSeverity } from '@shared/models/alarm.models'; |
|
|
|
import { entityFields } from '@shared/models/entity.models'; |
|
|
|
import { isDefinedAndNotNull, isUndefined } from '@core/utils'; |
|
|
|
import { deepClone, isDefinedAndNotNull, isUndefined } from '@core/utils'; |
|
|
|
import { CmdWrapper, WsService, WsSubscriber } from '@shared/models/websocket/websocket.models'; |
|
|
|
import { TelemetryWebsocketService } from '@core/ws/telemetry-websocket.service'; |
|
|
|
import { Notification, NotificationType } from '@shared/models/notification.models'; |
|
|
|
@ -141,6 +141,7 @@ export enum WsCmdType { |
|
|
|
ENTITY_COUNT = 'ENTITY_COUNT', |
|
|
|
ALARM_DATA = 'ALARM_DATA', |
|
|
|
ALARM_COUNT = 'ALARM_COUNT', |
|
|
|
ALARM_STATUS = 'ALARM_STATUS', |
|
|
|
|
|
|
|
NOTIFICATIONS = 'NOTIFICATIONS', |
|
|
|
NOTIFICATIONS_COUNT = 'NOTIFICATIONS_COUNT', |
|
|
|
@ -149,6 +150,7 @@ export enum WsCmdType { |
|
|
|
|
|
|
|
ALARM_DATA_UNSUBSCRIBE = 'ALARM_DATA_UNSUBSCRIBE', |
|
|
|
ALARM_COUNT_UNSUBSCRIBE = 'ALARM_COUNT_UNSUBSCRIBE', |
|
|
|
ALARM_STATUS_UNSUBSCRIBE = 'ALARM_STATUS_UNSUBSCRIBE', |
|
|
|
ENTITY_DATA_UNSUBSCRIBE = 'ENTITY_DATA_UNSUBSCRIBE', |
|
|
|
ENTITY_COUNT_UNSUBSCRIBE = 'ENTITY_COUNT_UNSUBSCRIBE', |
|
|
|
NOTIFICATIONS_UNSUBSCRIBE = 'NOTIFICATIONS_UNSUBSCRIBE' |
|
|
|
@ -298,6 +300,14 @@ export class AlarmCountCmd implements WebsocketCmd { |
|
|
|
type = WsCmdType.ALARM_COUNT; |
|
|
|
} |
|
|
|
|
|
|
|
export class AlarmStatusCmd implements WebsocketCmd { |
|
|
|
cmdId: number; |
|
|
|
originatorId: EntityId; |
|
|
|
severityList?: Array<AlarmSeverity>; |
|
|
|
typeList?: Array<string>; |
|
|
|
type = WsCmdType.ALARM_STATUS; |
|
|
|
} |
|
|
|
|
|
|
|
export class UnreadCountSubCmd implements WebsocketCmd { |
|
|
|
cmdId: number; |
|
|
|
type = WsCmdType.NOTIFICATIONS_COUNT; |
|
|
|
@ -352,6 +362,11 @@ export class AlarmCountUnsubscribeCmd implements WebsocketCmd { |
|
|
|
type = WsCmdType.ALARM_COUNT_UNSUBSCRIBE; |
|
|
|
} |
|
|
|
|
|
|
|
export class AlarmStatusUnsubscribeCmd implements WebsocketCmd { |
|
|
|
cmdId: number; |
|
|
|
type = WsCmdType.ALARM_STATUS_UNSUBSCRIBE; |
|
|
|
} |
|
|
|
|
|
|
|
export class UnsubscribeCmd implements WebsocketCmd { |
|
|
|
cmdId: number; |
|
|
|
type = WsCmdType.NOTIFICATIONS_UNSUBSCRIBE; |
|
|
|
@ -432,6 +447,7 @@ export enum CmdUpdateType { |
|
|
|
ENTITY_DATA = 'ENTITY_DATA', |
|
|
|
ALARM_DATA = 'ALARM_DATA', |
|
|
|
ALARM_COUNT_DATA = 'ALARM_COUNT_DATA', |
|
|
|
ALARM_STATUS = 'ALARM_STATUS', |
|
|
|
COUNT_DATA = 'COUNT_DATA', |
|
|
|
NOTIFICATIONS_COUNT = 'NOTIFICATIONS_COUNT', |
|
|
|
NOTIFICATIONS = 'NOTIFICATIONS' |
|
|
|
@ -469,6 +485,11 @@ export interface AlarmCountUpdateMsg extends CmdUpdateMsg { |
|
|
|
count: number; |
|
|
|
} |
|
|
|
|
|
|
|
export interface AlarmStatusUpdateMsg extends CmdUpdateMsg { |
|
|
|
cmdUpdateType: CmdUpdateType.ALARM_STATUS; |
|
|
|
active: boolean; |
|
|
|
} |
|
|
|
|
|
|
|
export interface NotificationCountUpdateMsg extends CmdUpdateMsg { |
|
|
|
cmdUpdateType: CmdUpdateType.NOTIFICATIONS_COUNT; |
|
|
|
totalUnreadCount: number; |
|
|
|
@ -506,6 +527,11 @@ export const isAlarmCountUpdateMsg = (message: WebsocketDataMsg): message is Ala |
|
|
|
return updateMsg.cmdId !== undefined && updateMsg.cmdUpdateType === CmdUpdateType.ALARM_COUNT_DATA; |
|
|
|
}; |
|
|
|
|
|
|
|
export const isAlarmStatusUpdateMsg = (message: WebsocketDataMsg): message is AlarmCountUpdateMsg => { |
|
|
|
const updateMsg = (message as CmdUpdateMsg); |
|
|
|
return updateMsg.cmdId !== undefined && updateMsg.cmdUpdateType === CmdUpdateType.ALARM_STATUS; |
|
|
|
}; |
|
|
|
|
|
|
|
export const isNotificationCountUpdateMsg = (message: WebsocketDataMsg): message is NotificationCountUpdateMsg => { |
|
|
|
const updateMsg = (message as CmdUpdateMsg); |
|
|
|
return updateMsg.cmdId !== undefined && updateMsg.cmdUpdateType === CmdUpdateType.NOTIFICATIONS_COUNT; |
|
|
|
@ -705,6 +731,15 @@ export class AlarmCountUpdate extends CmdUpdate { |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
export class AlarmStatusUpdate extends CmdUpdate { |
|
|
|
active: boolean; |
|
|
|
|
|
|
|
constructor(msg: AlarmStatusUpdateMsg) { |
|
|
|
super(msg); |
|
|
|
this.active = msg.active; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
export class NotificationCountUpdate extends CmdUpdate { |
|
|
|
totalUnreadCount: number; |
|
|
|
sequenceNumber: number; |
|
|
|
@ -750,14 +785,29 @@ export class SharedTelemetrySubscriber { |
|
|
|
return key; |
|
|
|
} |
|
|
|
|
|
|
|
private static createAlarmStatusSubscriberKey (entityId: EntityId, severityList: AlarmSeverity[] = null, typeList: string[] = null): string { |
|
|
|
let key = entityId.entityType + '_' + entityId.id; |
|
|
|
if (severityList) { |
|
|
|
key += '_' + severityList.sort().join('_'); |
|
|
|
} |
|
|
|
if (typeList) { |
|
|
|
key += '_' + typeList.sort().join('_'); |
|
|
|
} |
|
|
|
return key; |
|
|
|
} |
|
|
|
|
|
|
|
private subscribed = false; |
|
|
|
|
|
|
|
private attributeDataSubject = connectable(this.sharedSubscriptionInfo.subscriber.attributeData$(), |
|
|
|
{ connector: () => new ReplaySubject<Array<AttributeData>>(1)}); |
|
|
|
|
|
|
|
private alarmStatusSubject = connectable(this.sharedSubscriptionInfo.subscriber.alarmStatus$, |
|
|
|
{ connector: () => new ReplaySubject<AlarmStatusUpdate>()}); |
|
|
|
|
|
|
|
private subscriptions = new Array<Subscription>(); |
|
|
|
|
|
|
|
public attributeData$: Observable<Array<AttributeData>> = this.attributeDataSubject; //this.attributeDataSubject.asObservable();
|
|
|
|
public alarmStatus$: Observable<AlarmStatusUpdate> = this.alarmStatusSubject; |
|
|
|
|
|
|
|
public static createEntityAttributesSubscription(telemetryService: TelemetryWebsocketService, |
|
|
|
entityId: EntityId, attributeScope: TelemetryType, |
|
|
|
@ -781,6 +831,28 @@ export class SharedTelemetrySubscriber { |
|
|
|
return sharedSubscriber; |
|
|
|
} |
|
|
|
|
|
|
|
public static createAlarmStatusSubscription(telemetryService: TelemetryWebsocketService, |
|
|
|
entityId: EntityId, zone: NgZone, severityList: AlarmSeverity[] = null, |
|
|
|
typeList: string[] = null): SharedTelemetrySubscriber { |
|
|
|
const key = SharedTelemetrySubscriber.createAlarmStatusSubscriberKey(entityId, severityList, typeList); |
|
|
|
let info = SharedTelemetrySubscriber.subscribersCache[key]; |
|
|
|
if (!info) { |
|
|
|
const subscriber = TelemetrySubscriber.createAlarmStatusSubscription( |
|
|
|
telemetryService, entityId, zone, severityList, typeList |
|
|
|
); |
|
|
|
info = { |
|
|
|
key, |
|
|
|
subscriber, |
|
|
|
subscribed: false, |
|
|
|
sharedSubscribers: new Set<SharedTelemetrySubscriber>() |
|
|
|
}; |
|
|
|
SharedTelemetrySubscriber.subscribersCache[key] = info; |
|
|
|
} |
|
|
|
const sharedSubscriber = new SharedTelemetrySubscriber(info); |
|
|
|
info.sharedSubscribers.add(sharedSubscriber); |
|
|
|
return sharedSubscriber; |
|
|
|
} |
|
|
|
|
|
|
|
private constructor(private sharedSubscriptionInfo: SharedSubscriptionInfo) { |
|
|
|
} |
|
|
|
|
|
|
|
@ -788,6 +860,7 @@ export class SharedTelemetrySubscriber { |
|
|
|
if (!this.subscribed) { |
|
|
|
this.subscribed = true; |
|
|
|
this.subscriptions.push(this.attributeDataSubject.connect()); |
|
|
|
this.subscriptions.push(this.alarmStatusSubject.connect()); |
|
|
|
if (!this.sharedSubscriptionInfo.subscribed) { |
|
|
|
this.sharedSubscriptionInfo.subscriber.subscribe(); |
|
|
|
this.sharedSubscriptionInfo.subscribed = true; |
|
|
|
@ -823,6 +896,7 @@ export class TelemetrySubscriber extends WsSubscriber { |
|
|
|
private alarmDataSubject = new ReplaySubject<AlarmDataUpdate>(1); |
|
|
|
private entityCountSubject = new ReplaySubject<EntityCountUpdate>(1); |
|
|
|
private alarmCountSubject = new ReplaySubject<AlarmCountUpdate>(1); |
|
|
|
private alarmStatusSubject = new ReplaySubject<AlarmStatusUpdate>(1); |
|
|
|
private tsOffset = undefined; |
|
|
|
|
|
|
|
public data$ = this.dataSubject.asObservable(); |
|
|
|
@ -830,6 +904,7 @@ export class TelemetrySubscriber extends WsSubscriber { |
|
|
|
public alarmData$ = this.alarmDataSubject.asObservable(); |
|
|
|
public entityCount$ = this.entityCountSubject.asObservable(); |
|
|
|
public alarmCount$ = this.alarmCountSubject.asObservable(); |
|
|
|
public alarmStatus$ = this.alarmStatusSubject.asObservable(); |
|
|
|
|
|
|
|
public static createEntityAttributesSubscription(telemetryService: TelemetryWebsocketService, |
|
|
|
entityId: EntityId, attributeScope: TelemetryType, |
|
|
|
@ -851,6 +926,17 @@ export class TelemetrySubscriber extends WsSubscriber { |
|
|
|
return subscriber; |
|
|
|
} |
|
|
|
|
|
|
|
public static createAlarmStatusSubscription(telemetryService: TelemetryWebsocketService, entityId: EntityId, |
|
|
|
zone: NgZone, severityList: AlarmSeverity[] = null, typeList: string[] = null): TelemetrySubscriber { |
|
|
|
const subscriptionCommand = new AlarmStatusCmd(); |
|
|
|
subscriptionCommand.originatorId = deepClone(entityId); |
|
|
|
subscriptionCommand.severityList = severityList; |
|
|
|
subscriptionCommand.typeList = typeList; |
|
|
|
const subscriber = new TelemetrySubscriber(telemetryService, zone); |
|
|
|
subscriber.subscriptionCommands.push(subscriptionCommand); |
|
|
|
return subscriber; |
|
|
|
} |
|
|
|
|
|
|
|
public static createEntityFilterLatestSubscription(telemetryService: TelemetryWebsocketService, |
|
|
|
entityFilter: EntityFilter, zone: NgZone, |
|
|
|
latestKeys: EntityKey[] = null): TelemetrySubscriber { |
|
|
|
@ -882,6 +968,7 @@ export class TelemetrySubscriber extends WsSubscriber { |
|
|
|
this.alarmDataSubject.complete(); |
|
|
|
this.entityCountSubject.complete(); |
|
|
|
this.alarmCountSubject.complete(); |
|
|
|
this.alarmStatusSubject.complete(); |
|
|
|
super.complete(); |
|
|
|
} |
|
|
|
|
|
|
|
@ -971,6 +1058,18 @@ export class TelemetrySubscriber extends WsSubscriber { |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
public onAlarmStatus(message: AlarmStatusUpdate) { |
|
|
|
if (this.zone) { |
|
|
|
this.zone.run( |
|
|
|
() => { |
|
|
|
this.alarmStatusSubject.next(message); |
|
|
|
} |
|
|
|
); |
|
|
|
} else { |
|
|
|
this.alarmStatusSubject.next(message); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
public attributeData$(): Observable<Array<AttributeData>> { |
|
|
|
const attributeData = new Array<AttributeData>(); |
|
|
|
return this.data$.pipe( |
|
|
|
|
Andrew Shvayka
2 years ago
GitHub