github/kubeshark
1
0
Fork 0
mirror of https://github.com/kubeshark/kubeshark.git synced 2025-08-11 13:23:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update socket_server_handlers.go

Browse Source
This commit is contained in:
Rami Berman 2022-04-03 13:35:20 +03:00
parent e537e8fbd6
commit b568f42c59
1 changed files with 4 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
agent/pkg/api/socket_server_handlers.go
View File

@ -98,11 +98,10 @@ func BroadcastToTapperClients(message []byte) {
func (h *RoutesEventHandlers) WebSocketMessage(socketId int, isTapper bool, message []byte) { func (h *RoutesEventHandlers) WebSocketMessage(socketId int, isTapper bool, message []byte) {
if isTapper { if isTapper {
h.handleTapperIncomingMessage(message) HandleTapperIncomingMessage(message, h.SocketOutChannel, BroadcastToBrowserClients)
} else { } else {
// we initiate the basenine stream after the first websocket message we receive (it contains the entry query), we then store a cancelfunc to later cancel this stream // we initiate the basenine stream after the first websocket message we receive (it contains the entry query), we then store a cancelfunc to later cancel this stream
if browserClients[socketId] != nil && browserClients[socketId].dataStreamCancelFunc == nil { if browserClients[socketId] != nil && browserClients[socketId].dataStreamCancelFunc == nil {
var params WebSocketParams var params WebSocketParams
if err := json.Unmarshal(message, &params); err != nil { if err := json.Unmarshal(message, &params); err != nil {
logger.Log.Errorf("Error: %v", socketId, err) logger.Log.Errorf("Error: %v", socketId, err)
@ -121,7 +120,7 @@ func (h *RoutesEventHandlers) WebSocketMessage(socketId int, isTapper bool, mess
} }
} }
func (h *RoutesEventHandlers) handleTapperIncomingMessage(message []byte) { func HandleTapperIncomingMessage(message []byte, socketOutChannel chan<- *tapApi.OutputChannelItem, broadcastMessageFunc func([]byte)) {
var socketMessageBase shared.WebSocketMessageMetadata var socketMessageBase shared.WebSocketMessageMetadata
err := json.Unmarshal(message, &socketMessageBase) err := json.Unmarshal(message, &socketMessageBase)
if err != nil { if err != nil {
@ -135,7 +134,7 @@ func (h *RoutesEventHandlers) handleTapperIncomingMessage(message []byte) {
logger.Log.Infof("Could not unmarshal message of message type %s %v", socketMessageBase.MessageType, err) logger.Log.Infof("Could not unmarshal message of message type %s %v", socketMessageBase.MessageType, err)
} else { } else {
// NOTE: This is where the message comes back from the intermediate WebSocket to code. // NOTE: This is where the message comes back from the intermediate WebSocket to code.
h.SocketOutChannel <- tappedEntryMessage.Data socketOutChannel <- tappedEntryMessage.Data
} }
case shared.WebSocketMessageTypeUpdateStatus: case shared.WebSocketMessageTypeUpdateStatus:
var statusMessage shared.WebSocketStatusMessage var statusMessage shared.WebSocketStatusMessage
@ -143,7 +142,7 @@ func (h *RoutesEventHandlers) handleTapperIncomingMessage(message []byte) {
if err != nil { if err != nil {
logger.Log.Infof("Could not unmarshal message of message type %s %v", socketMessageBase.MessageType, err) logger.Log.Infof("Could not unmarshal message of message type %s %v", socketMessageBase.MessageType, err)
} else { } else {
BroadcastToBrowserClients(message) broadcastMessageFunc(message)
} }
default: default:
logger.Log.Infof("Received socket message of type %s for which no handlers are defined", socketMessageBase.MessageType) logger.Log.Infof("Received socket message of type %s for which no handlers are defined", socketMessageBase.MessageType)