Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 1 | /* |
| 2 | * Copyright 2020-present Open Networking Foundation |
| 3 | * |
| 4 | * Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | * you may not use this file except in compliance with the License. |
| 6 | * You may obtain a copy of the License at |
| 7 | * |
| 8 | * http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | * |
| 10 | * Unless required by applicable law or agreed to in writing, software |
| 11 | * distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | * See the License for the specific language governing permissions and |
| 14 | * limitations under the License. |
| 15 | */ |
| 16 | |
| 17 | package event |
| 18 | |
| 19 | import ( |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 20 | "context" |
Andrea Campanella | 4afb2f0 | 2021-01-29 09:38:57 +0100 | [diff] [blame] | 21 | "encoding/binary" |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 22 | "encoding/hex" |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 23 | "fmt" |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 24 | "github.com/golang/protobuf/ptypes/empty" |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 25 | "github.com/opencord/voltha-go/rw_core/utils" |
| 26 | "github.com/opencord/voltha-lib-go/v4/pkg/events/eventif" |
Maninder | dfadc98 | 2020-10-28 14:04:33 +0530 | [diff] [blame] | 27 | "github.com/opencord/voltha-lib-go/v4/pkg/log" |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 28 | "github.com/opencord/voltha-protos/v4/go/common" |
Maninder | dfadc98 | 2020-10-28 14:04:33 +0530 | [diff] [blame] | 29 | "github.com/opencord/voltha-protos/v4/go/openflow_13" |
| 30 | "github.com/opencord/voltha-protos/v4/go/voltha" |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 31 | "github.com/opentracing/opentracing-go" |
| 32 | jtracing "github.com/uber/jaeger-client-go" |
| 33 | "sync" |
| 34 | "time" |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 35 | ) |
| 36 | |
| 37 | type Manager struct { |
| 38 | packetInQueue chan openflow_13.PacketIn |
| 39 | packetInQueueDone chan bool |
| 40 | changeEventQueue chan openflow_13.ChangeEvent |
| 41 | changeEventQueueDone chan bool |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 42 | RPCEventManager *RPCEventManager |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 43 | } |
| 44 | |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 45 | type RPCEventManager struct { |
| 46 | eventProxy eventif.EventProxy |
| 47 | coreInstanceID string |
Himani Chawla | 9cfc499 | 2021-03-22 12:43:01 +0530 | [diff] [blame^] | 48 | stackID string |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 49 | } |
| 50 | |
Himani Chawla | 9cfc499 | 2021-03-22 12:43:01 +0530 | [diff] [blame^] | 51 | func NewManager(proxyForRPCEvents eventif.EventProxy, instanceID string, stackID string) *Manager { |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 52 | return &Manager{ |
| 53 | packetInQueue: make(chan openflow_13.PacketIn, 100), |
| 54 | packetInQueueDone: make(chan bool, 1), |
| 55 | changeEventQueue: make(chan openflow_13.ChangeEvent, 100), |
| 56 | changeEventQueueDone: make(chan bool, 1), |
Himani Chawla | 9cfc499 | 2021-03-22 12:43:01 +0530 | [diff] [blame^] | 57 | RPCEventManager: NewRPCEventManager(proxyForRPCEvents, instanceID, stackID), |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 58 | } |
| 59 | } |
| 60 | |
Himani Chawla | 9cfc499 | 2021-03-22 12:43:01 +0530 | [diff] [blame^] | 61 | func NewRPCEventManager(proxyForRPCEvents eventif.EventProxy, instanceID string, stackID string) *RPCEventManager { |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 62 | return &RPCEventManager{ |
| 63 | eventProxy: proxyForRPCEvents, |
| 64 | coreInstanceID: instanceID, |
Himani Chawla | 9cfc499 | 2021-03-22 12:43:01 +0530 | [diff] [blame^] | 65 | stackID: stackID, |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 66 | } |
| 67 | } |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 68 | func (q *Manager) SendPacketIn(ctx context.Context, deviceID string, transationID string, packet *openflow_13.OfpPacketIn) { |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 69 | // TODO: Augment the OF PacketIn to include the transactionId |
| 70 | packetIn := openflow_13.PacketIn{Id: deviceID, PacketIn: packet} |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 71 | logger.Debugw(ctx, "send-packet-in", log.Fields{"packet-in": packetIn}) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 72 | q.packetInQueue <- packetIn |
| 73 | } |
| 74 | |
| 75 | type callTracker struct { |
| 76 | failedPacket interface{} |
| 77 | } |
| 78 | type streamTracker struct { |
| 79 | calls map[string]*callTracker |
| 80 | sync.Mutex |
| 81 | } |
| 82 | |
| 83 | var streamingTracker = &streamTracker{calls: make(map[string]*callTracker)} |
| 84 | |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 85 | func (q *Manager) getStreamingTracker(ctx context.Context, method string, done chan<- bool) *callTracker { |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 86 | streamingTracker.Lock() |
| 87 | defer streamingTracker.Unlock() |
| 88 | if _, ok := streamingTracker.calls[method]; ok { |
| 89 | // bail out the other packet in thread |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 90 | logger.Debugf(ctx, "%s-streaming-call-already-running-exiting-it", method) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 91 | done <- true |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 92 | logger.Debugf(ctx, "last-%s-exited-continuing", method) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 93 | } else { |
| 94 | streamingTracker.calls[method] = &callTracker{failedPacket: nil} |
| 95 | } |
| 96 | return streamingTracker.calls[method] |
| 97 | } |
| 98 | |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 99 | func (q *Manager) flushFailedPackets(ctx context.Context, tracker *callTracker) error { |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 100 | if tracker.failedPacket != nil { |
| 101 | switch tracker.failedPacket.(type) { |
| 102 | case openflow_13.PacketIn: |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 103 | logger.Debug(ctx, "enqueueing-last-failed-packet-in") |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 104 | q.packetInQueue <- tracker.failedPacket.(openflow_13.PacketIn) |
| 105 | case openflow_13.ChangeEvent: |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 106 | logger.Debug(ctx, "enqueueing-last-failed-change-event") |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 107 | q.changeEventQueue <- tracker.failedPacket.(openflow_13.ChangeEvent) |
| 108 | } |
| 109 | } |
| 110 | return nil |
| 111 | } |
| 112 | |
| 113 | // ReceivePacketsIn receives packets from adapter |
| 114 | func (q *Manager) ReceivePacketsIn(_ *empty.Empty, packetsIn voltha.VolthaService_ReceivePacketsInServer) error { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 115 | ctx := context.Background() |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 116 | ctx = utils.WithRPCMetadataContext(ctx, "ReceivePacketsIn") |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 117 | var streamingTracker = q.getStreamingTracker(ctx, "ReceivePacketsIn", q.packetInQueueDone) |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 118 | logger.Debugw(ctx, "receive-packets-in-request", log.Fields{"packets-in": packetsIn}) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 119 | |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 120 | err := q.flushFailedPackets(ctx, streamingTracker) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 121 | if err != nil { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 122 | logger.Errorw(ctx, "unable-to-flush-failed-packets", log.Fields{"error": err}) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 123 | } |
| 124 | |
| 125 | loop: |
| 126 | for { |
| 127 | select { |
| 128 | case packet := <-q.packetInQueue: |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 129 | logger.Debugw(ctx, "sending-packet-in", log.Fields{ |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 130 | "packet": hex.EncodeToString(packet.PacketIn.Data), |
| 131 | }) |
| 132 | if err := packetsIn.Send(&packet); err != nil { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 133 | logger.Errorw(ctx, "failed-to-send-packet", log.Fields{"error": err}) |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 134 | go q.RPCEventManager.GetAndSendRPCEvent(ctx, packet.Id, err.Error(), |
| 135 | nil, "RPC_ERROR_RAISE_EVENT", voltha.EventCategory_COMMUNICATION, |
| 136 | nil, time.Now().UnixNano()) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 137 | // save the last failed packet in |
| 138 | streamingTracker.failedPacket = packet |
| 139 | } else { |
| 140 | if streamingTracker.failedPacket != nil { |
| 141 | // reset last failed packet saved to avoid flush |
| 142 | streamingTracker.failedPacket = nil |
| 143 | } |
| 144 | } |
| 145 | case <-q.packetInQueueDone: |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 146 | logger.Debug(ctx, "another-receive-packets-in-running-bailing-out") |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 147 | break loop |
| 148 | } |
| 149 | } |
| 150 | |
| 151 | //TODO: Find an elegant way to get out of the above loop when the Core is stopped |
| 152 | return nil |
| 153 | } |
| 154 | |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 155 | func (q *Manager) SendChangeEvent(ctx context.Context, deviceID string, reason openflow_13.OfpPortReason, desc *openflow_13.OfpPort) { |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 156 | logger.Debugw(ctx, "send-change-event", log.Fields{"device-id": deviceID, "reason": reason, "desc": desc}) |
Kent Hagerman | fa9d6d4 | 2020-05-25 11:49:40 -0400 | [diff] [blame] | 157 | q.changeEventQueue <- openflow_13.ChangeEvent{ |
| 158 | Id: deviceID, |
| 159 | Event: &openflow_13.ChangeEvent_PortStatus{ |
| 160 | PortStatus: &openflow_13.OfpPortStatus{ |
| 161 | Reason: reason, |
| 162 | Desc: desc, |
| 163 | }, |
| 164 | }, |
| 165 | } |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 166 | } |
| 167 | |
Andrea Campanella | 4afb2f0 | 2021-01-29 09:38:57 +0100 | [diff] [blame] | 168 | func (q *Manager) SendFlowChangeEvent(ctx context.Context, deviceID string, res []error, xid uint32, flowCookie uint64) { |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 169 | logger.Debugw(ctx, "send-change-event", log.Fields{"device-id": deviceID, |
| 170 | "flow-id": xid, "flow-cookie": flowCookie, "errors": res}) |
Maninder | f421da6 | 2020-12-04 11:44:58 +0530 | [diff] [blame] | 171 | errorType := openflow_13.OfpErrorType_OFPET_FLOW_MOD_FAILED |
Andrea Campanella | 4afb2f0 | 2021-01-29 09:38:57 +0100 | [diff] [blame] | 172 | //Manually creating the data payload for the flow error message |
| 173 | bs := make([]byte, 2) |
| 174 | //OF 1.3 |
| 175 | bs[0] = byte(4) |
| 176 | //Flow Mod |
| 177 | bs[1] = byte(14) |
| 178 | //Length of the message |
| 179 | length := make([]byte, 2) |
| 180 | binary.BigEndian.PutUint16(length, 56) |
| 181 | bs = append(bs, length...) |
| 182 | emptyArr := []byte{0, 0, 0, 0} |
| 183 | bs = append(bs, emptyArr...) |
| 184 | //Cookie of the Flow |
| 185 | cookie := make([]byte, 52) |
| 186 | binary.BigEndian.PutUint64(cookie, flowCookie) |
| 187 | bs = append(bs, cookie...) |
Maninder | f421da6 | 2020-12-04 11:44:58 +0530 | [diff] [blame] | 188 | q.changeEventQueue <- openflow_13.ChangeEvent{ |
| 189 | Id: deviceID, |
| 190 | Event: &openflow_13.ChangeEvent_Error{ |
| 191 | Error: &openflow_13.OfpErrorMsg{ |
| 192 | Header: &openflow_13.OfpHeader{ |
Andrea Campanella | 4afb2f0 | 2021-01-29 09:38:57 +0100 | [diff] [blame] | 193 | Type: openflow_13.OfpType_OFPT_FLOW_MOD, |
| 194 | Xid: xid, |
Maninder | f421da6 | 2020-12-04 11:44:58 +0530 | [diff] [blame] | 195 | }, |
| 196 | Type: uint32(errorType), |
| 197 | Code: uint32(openflow_13.OfpFlowModFailedCode_OFPFMFC_UNKNOWN), |
Andrea Campanella | 4afb2f0 | 2021-01-29 09:38:57 +0100 | [diff] [blame] | 198 | Data: bs, |
Maninder | f421da6 | 2020-12-04 11:44:58 +0530 | [diff] [blame] | 199 | }, |
| 200 | }, |
| 201 | } |
| 202 | } |
| 203 | |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 204 | // ReceiveChangeEvents receives change in events |
| 205 | func (q *Manager) ReceiveChangeEvents(_ *empty.Empty, changeEvents voltha.VolthaService_ReceiveChangeEventsServer) error { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 206 | ctx := context.Background() |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 207 | ctx = utils.WithRPCMetadataContext(ctx, "ReceiveChangeEvents") |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 208 | var streamingTracker = q.getStreamingTracker(ctx, "ReceiveChangeEvents", q.changeEventQueueDone) |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 209 | logger.Debugw(ctx, "receive-change-events-request", log.Fields{"change-events": changeEvents}) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 210 | |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 211 | err := q.flushFailedPackets(ctx, streamingTracker) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 212 | if err != nil { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 213 | logger.Errorw(ctx, "unable-to-flush-failed-packets", log.Fields{"error": err}) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 214 | } |
| 215 | |
| 216 | loop: |
| 217 | for { |
| 218 | select { |
| 219 | // Dequeue a change event |
| 220 | case event := <-q.changeEventQueue: |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 221 | logger.Debugw(ctx, "sending-change-event", log.Fields{"event": event}) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 222 | if err := changeEvents.Send(&event); err != nil { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 223 | logger.Errorw(ctx, "failed-to-send-change-event", log.Fields{"error": err}) |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 224 | go q.RPCEventManager.GetAndSendRPCEvent(ctx, event.Id, err.Error(), |
| 225 | nil, "RPC_ERROR_RAISE_EVENT", voltha.EventCategory_COMMUNICATION, nil, |
| 226 | time.Now().UnixNano()) |
| 227 | // save last failed change event |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 228 | streamingTracker.failedPacket = event |
| 229 | } else { |
| 230 | if streamingTracker.failedPacket != nil { |
| 231 | // reset last failed event saved on success to avoid flushing |
| 232 | streamingTracker.failedPacket = nil |
| 233 | } |
| 234 | } |
| 235 | case <-q.changeEventQueueDone: |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 236 | logger.Debug(ctx, "another-receive-change-events-already-running-bailing-out") |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 237 | break loop |
| 238 | } |
| 239 | } |
| 240 | |
| 241 | return nil |
| 242 | } |
| 243 | |
| 244 | func (q *Manager) GetChangeEventsQueueForTest() <-chan openflow_13.ChangeEvent { |
| 245 | return q.changeEventQueue |
| 246 | } |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 247 | |
| 248 | func (q *RPCEventManager) NewRPCEvent(ctx context.Context, resourceID, desc string, context map[string]string) *voltha.RPCEvent { |
| 249 | logger.Debugw(ctx, "new-rpc-event", log.Fields{"resource-id": resourceID}) |
| 250 | var opID string |
| 251 | var rpc string |
| 252 | |
| 253 | if span := opentracing.SpanFromContext(ctx); span != nil { |
| 254 | if jSpan, ok := span.(*jtracing.Span); ok { |
| 255 | opID = fmt.Sprintf("%016x", jSpan.SpanContext().TraceID().Low) // Using Sprintf to avoid removal of leading 0s |
| 256 | } |
| 257 | } |
| 258 | rpc = utils.GetRPCMetadataFromContext(ctx) |
| 259 | rpcev := &voltha.RPCEvent{ |
| 260 | Rpc: rpc, |
| 261 | OperationId: opID, |
| 262 | ResourceId: resourceID, |
| 263 | Service: q.coreInstanceID, |
Himani Chawla | 9cfc499 | 2021-03-22 12:43:01 +0530 | [diff] [blame^] | 264 | StackId: q.stackID, |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 265 | Status: &common.OperationResp{ |
| 266 | Code: common.OperationResp_OPERATION_FAILURE, |
| 267 | }, |
| 268 | Description: desc, |
| 269 | Context: context, |
| 270 | } |
| 271 | return rpcev |
| 272 | } |
| 273 | |
| 274 | func (q *RPCEventManager) SendRPCEvent(ctx context.Context, id string, rpcEvent *voltha.RPCEvent, category voltha.EventCategory_Types, subCategory *voltha.EventSubCategory_Types, raisedTs int64) { |
| 275 | //TODO Instead of directly sending to the kafka bus, queue the message and send it asynchronously |
| 276 | if rpcEvent.Rpc != "" { |
| 277 | _ = q.eventProxy.SendRPCEvent(ctx, id, rpcEvent, category, subCategory, raisedTs) |
| 278 | } |
| 279 | } |
| 280 | |
| 281 | func (q *RPCEventManager) GetAndSendRPCEvent(ctx context.Context, resourceID, desc string, context map[string]string, |
| 282 | id string, category voltha.EventCategory_Types, subCategory *voltha.EventSubCategory_Types, raisedTs int64) { |
| 283 | rpcEvent := q.NewRPCEvent(ctx, resourceID, desc, context) |
| 284 | //TODO Instead of directly sending to the kafka bus, queue the message and send it asynchronously |
| 285 | if rpcEvent.Rpc != "" { |
| 286 | _ = q.eventProxy.SendRPCEvent(ctx, id, rpcEvent, category, subCategory, raisedTs) |
| 287 | } |
| 288 | } |