@ -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 (
Max Petrov
2 years ago
GitHub
