Implement Event Subscription handling for channels #746
Conversation
|
Hello, I was trying to use your implementation of event subscription but i am getting error- "The attribute is not supported for the specified Node. StatusBadAttributeIDInvalid (0x80350000)". I am subscribing to an object node which has below configuration- I have tested this node on UA Expert and there i am able to see the events and alarms. I believe i need to change some configuration to subscribe events on this node, could you help with it @Jwcode-uk ? |
|
Thank you @Jwcode-uk. I can see that one function in Not sure if this is complex enough to use the // Deprectated: Use NewDefaultMonitoredItemCreateRequest instead. Will be removed with 0.8.0
func NewMonitoredItemCreateRequestWithDefaults(nodeID *ua.NodeID, attributeID ua.AttributeID, clientHandle uint32) *ua.MonitoredItemCreateRequest {
return NewDefaultMonitoredItemCreateRequest(MICreateRequestArgs{
NodeID: nodeID,
AttributeID: attributeID,
ClientHandle: clientHandle,
})
}
type MonitoredItemCreateRequestArgs struct {
NodeID *ua.NodeID
AttributeID ua.AttributeID
ClientHandle uint32
Filter *ua.ExtensionObject
}
func NewDefaultMonitoredItemCreateRequest(args MonitoredItemCreateRequestArgs) *ua.MonitoredItemCreateRequest {
if args.AttributeID == 0 {
args.AttributeID = ua.AttributeIDValue
}
return &ua.MonitoredItemCreateRequest{
ItemToMonitor: &ua.ReadValueID{
NodeID: args.NodeID,
AttributeID: args.AttributeID,
DataEncoding: &ua.QualifiedName{},
},
MonitoringMode: ua.MonitoringModeReporting,
RequestedParameters: &ua.MonitoringParameters{
ClientHandle: args.ClientHandle,
DiscardOldest: true,
Filter: args.Filter,
QueueSize: 10,
SamplingInterval: 0.0,
},
}
} |
Signed-off-by: Jack Chen <[email protected]>
jackchenjc
force-pushed
the
DAN-11-branch
branch
from
aa706a4 to
669ad21
Compare
There was a problem hiding this comment.
Did a first pass through the code and would like to see a change so that this implementation does not break existing code. See the comment on ChanSubscribe for a suggestion. I've also left some smaller style comments.
Some more comments would be useful since it isn't immediately obvious to me what the code does.
Lets do the refactor first - unless that isn't possible - and then I'll do another pass over the implementation and probably have a couple more questions.
| dataValue := &ua.DataValue{ | ||
| Value: field, | ||
| Status: ua.StatusOK, | ||
| SourceTimestamp: time.Now(), |
There was a problem hiding this comment.
Shouldn't all dataValues have the same timestamp? Isn't there a timestamp in the item (e.g. ServerTimestamp)?
| @@ -155,7 +168,7 @@ func (s *Subscription) sendError(err error) { | |||
| } | |||
|
|
|||
| // internal func to read from internal channel and write to client provided channel | |||
| func (s *Subscription) pump(ctx context.Context, notifyCh chan<- *DataChangeMessage, cb MsgHandler) { | |||
| func (s *Subscription) pump(ctx context.Context, notifyCh chan<- Message, cb MsgHandler, sub bool) { | |||
There was a problem hiding this comment.
sub seems unused here and the name also collides with other usages of sub in this file where it usually denotes a *Subscription
| @@ -254,16 +310,16 @@ func (s *Subscription) Dropped() uint64 { | |||
| } | |||
|
|
|||
| // AddNodes adds nodes defined by their string representation | |||
| func (s *Subscription) AddNodes(ctx context.Context, nodes ...string) error { | |||
| func (s *Subscription) AddNodes(ctx context.Context, eventSub bool, filter *ua.ExtensionObject, nodes ...string) error { | |||
There was a problem hiding this comment.
What does eventSub do here? Also, see comment on ChanSubscribe for a way to add this functionality without breaking the signatures.
| return nil, nil | ||
| } | ||
|
|
||
| toAdd := make([]*ua.MonitoredItemCreateRequest, 0) |
There was a problem hiding this comment.
var toAdd []*ua.MonitoredItemCreateRequest| out.EventFields = make([]*ua.DataValue, len(item.EventFields)) | ||
|
|
||
| for i, field := range item.EventFields { | ||
| // Create a new DataValue |
There was a problem hiding this comment.
pls remove this comment as it does not have value
| } | ||
| } | ||
|
|
||
| if notifyCh != nil { |
There was a problem hiding this comment.
I prefer switch statements over if/else, e.g.
switch {
case notifyCh != nil:
select { ... }
case cb != nil:
cb(sb, out)
atomic.AddUint64(&s.delivered, 1)
default:
panic(...)
}| DataEncoding: &ua.QualifiedName{}, | ||
| }, | ||
| MonitoringMode: ua.MonitoringModeReporting, | ||
| RequestedParameters: &ua.MonitoringParameters{ | ||
| ClientHandle: clientHandle, | ||
| ClientHandle: args.ClientHandle, |
There was a problem hiding this comment.
Shouldn't the other parameters be configurable as well while providing sensible defaults?
| @@ -137,13 +150,13 @@ func (m *NodeMonitor) Subscribe(ctx context.Context, params *opcua.SubscriptionP | |||
| // via the monitor's `ErrHandler`. | |||
| // The caller must call `Unsubscribe` to stop and clean up resources. Canceling the context | |||
| // will also cause the subscription to stop, but `Unsubscribe` must still be called. | |||
| func (m *NodeMonitor) ChanSubscribe(ctx context.Context, params *opcua.SubscriptionParameters, ch chan<- *DataChangeMessage, nodes ...string) (*Subscription, error) { | |||
| sub, err := newSubscription(ctx, m, params, 16, nodes...) | |||
| func (m *NodeMonitor) ChanSubscribe(ctx context.Context, params *opcua.SubscriptionParameters, ch chan<- Message, eventSub bool, filter *ua.ExtensionObject, nodes ...string) (*Subscription, error) { | |||
There was a problem hiding this comment.
I am starting to think that an args parameter might be useful here. We did something similar with the WithContext versions of all functions where we dropped the WithContext suffix in a later version. So maybe we can refactor this change so that we add ChanSubscribeWithArgs(ctx context.Context, args ChanSubscribeArgs) (*Subscription, error) and drop the WithArgs in 0.8.0. Add a deprecation note to the other version. What do you think?
Hi, I have added event channel handling. This is a breaking change as it changes the signatures of some of the functions to allow for handling of both event messages and data change messages. Thanks :)