iridium/neko
2
2
Fork 0
mirror of https://github.com/m1k1o/neko.git synced 2024-07-24 14:40:50 +12:00
Code Issues Packages Projects Releases Wiki Activity

fix naming convention.

Browse Source
This commit is contained in:
Miroslav Šedivý 2021-02-14 17:11:21 +01:00
parent 45679f1b86
commit 2b693f4e52
8 changed files with 142 additions and 142 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
internal/http/manager.go
View File

@ -78,24 +78,24 @@ func New(WebSocketManager types.WebSocketManager, ApiManager types.ApiManager, c
} }
} }
func (s *HttpManagerCtx) Start() { func (manager *HttpManagerCtx) Start() {
if s.conf.Cert != "" && s.conf.Key != "" { if manager.conf.Cert != "" && manager.conf.Key != "" {
go func() { go func() {
if err := s.http.ListenAndServeTLS(s.conf.Cert, s.conf.Key); err != http.ErrServerClosed { if err := manager.http.ListenAndServeTLS(manager.conf.Cert, manager.conf.Key); err != http.ErrServerClosed {
s.logger.Panic().Err(err).Msg("unable to start https server") manager.logger.Panic().Err(err).Msg("unable to start https server")
} }
}() }()
s.logger.Info().Msgf("https listening on %s", s.http.Addr) manager.logger.Info().Msgf("https listening on %s", manager.http.Addr)
} else { } else {
go func() { go func() {
if err := s.http.ListenAndServe(); err != http.ErrServerClosed { if err := manager.http.ListenAndServe(); err != http.ErrServerClosed {
s.logger.Panic().Err(err).Msg("unable to start http server") manager.logger.Panic().Err(err).Msg("unable to start http server")
} }
}() }()
s.logger.Info().Msgf("http listening on %s", s.http.Addr) manager.logger.Info().Msgf("http listening on %s", manager.http.Addr)
} }
} }
func (s *HttpManagerCtx) Shutdown() error { func (manager *HttpManagerCtx) Shutdown() error {
return s.http.Shutdown(context.Background()) return manager.http.Shutdown(context.Background())
} }

64
internal/session/session.go
View File

@ -9,14 +9,14 @@ import (
) )
type SessionCtx struct { type SessionCtx struct {
id string id string
logger zerolog.Logger logger zerolog.Logger
manager *SessionManagerCtx manager *SessionManagerCtx
profile types.MemberProfile profile types.MemberProfile
websocket_peer types.WebSocketPeer websocketPeer types.WebSocketPeer
websocket_connected bool websocketConnected bool
webrtc_peer types.WebRTCPeer webrtcPeer types.WebRTCPeer
webrtc_connected bool webrtcConnected bool
} }
func (session *SessionCtx) ID() string { func (session *SessionCtx) ID() string {
@ -77,7 +77,7 @@ func (session *SessionCtx) profileChanged() {
} }
if !session.CanWatch() && session.IsWatching() { if !session.CanWatch() && session.IsWatching() {
if err := session.webrtc_peer.Destroy(); err != nil { if err := session.webrtcPeer.Destroy(); err != nil {
session.logger.Warn().Err(err).Msgf("webrtc destroy has failed") session.logger.Warn().Err(err).Msgf("webrtc destroy has failed")
} }
} }
@ -98,11 +98,11 @@ func (session *SessionCtx) IsHost() bool {
} }
func (session *SessionCtx) IsConnected() bool { func (session *SessionCtx) IsConnected() bool {
return session.websocket_connected return session.websocketConnected
} }
func (session *SessionCtx) IsWatching() bool { func (session *SessionCtx) IsWatching() bool {
return session.webrtc_connected return session.webrtcConnected
} }
func (session *SessionCtx) GetState() types.MemberState { func (session *SessionCtx) GetState() types.MemberState {
@ -117,18 +117,18 @@ func (session *SessionCtx) GetState() types.MemberState {
// webscoket // webscoket
// --- // ---
func (session *SessionCtx) SetWebSocketPeer(websocket_peer types.WebSocketPeer) { func (session *SessionCtx) SetWebSocketPeer(websocketPeer types.WebSocketPeer) {
if session.websocket_peer != nil { if session.websocketPeer != nil {
if err := session.websocket_peer.Destroy(); err != nil { if err := session.websocketPeer.Destroy(); err != nil {
session.logger.Warn().Err(err).Msgf("websocket destroy has failed") session.logger.Warn().Err(err).Msgf("websocket destroy has failed")
} }
} }
session.websocket_peer = websocket_peer session.websocketPeer = websocketPeer
} }
func (session *SessionCtx) SetWebSocketConnected(connected bool) { func (session *SessionCtx) SetWebSocketConnected(connected bool) {
session.websocket_connected = connected session.websocketConnected = connected
if connected { if connected {
session.manager.emmiter.Emit("connected", session) session.manager.emmiter.Emit("connected", session)
@ -136,21 +136,21 @@ func (session *SessionCtx) SetWebSocketConnected(connected bool) {
} }
session.manager.emmiter.Emit("disconnected", session) session.manager.emmiter.Emit("disconnected", session)
session.websocket_peer = nil session.websocketPeer = nil
if session.webrtc_peer != nil { if session.webrtcPeer != nil {
if err := session.webrtc_peer.Destroy(); err != nil { if err := session.webrtcPeer.Destroy(); err != nil {
session.logger.Warn().Err(err).Msgf("webrtc destroy has failed") session.logger.Warn().Err(err).Msgf("webrtc destroy has failed")
} }
} }
} }
func (session *SessionCtx) Send(v interface{}) error { func (session *SessionCtx) Send(v interface{}) error {
if session.websocket_peer == nil { if session.websocketPeer == nil {
return nil return nil
} }
return session.websocket_peer.Send(v) return session.websocketPeer.Send(v)
} }
func (session *SessionCtx) Disconnect(reason string) error { func (session *SessionCtx) Disconnect(reason string) error {
@ -162,14 +162,14 @@ func (session *SessionCtx) Disconnect(reason string) error {
return err return err
} }
if session.websocket_peer != nil { if session.websocketPeer != nil {
if err := session.websocket_peer.Destroy(); err != nil { if err := session.websocketPeer.Destroy(); err != nil {
return err return err
} }
} }
if session.webrtc_peer != nil { if session.webrtcPeer != nil {
if err := session.webrtc_peer.Destroy(); err != nil { if err := session.webrtcPeer.Destroy(); err != nil {
return err return err
} }
} }
@ -181,25 +181,25 @@ func (session *SessionCtx) Disconnect(reason string) error {
// webrtc // webrtc
// --- // ---
func (session *SessionCtx) SetWebRTCPeer(webrtc_peer types.WebRTCPeer) { func (session *SessionCtx) SetWebRTCPeer(webrtcPeer types.WebRTCPeer) {
if session.webrtc_peer != nil { if session.webrtcPeer != nil {
if err := session.webrtc_peer.Destroy(); err != nil { if err := session.webrtcPeer.Destroy(); err != nil {
session.logger.Warn().Err(err).Msgf("webrtc destroy has failed") session.logger.Warn().Err(err).Msgf("webrtc destroy has failed")
} }
} }
session.webrtc_peer = webrtc_peer session.webrtcPeer = webrtcPeer
} }
func (session *SessionCtx) SetWebRTCConnected(connected bool) { func (session *SessionCtx) SetWebRTCConnected(connected bool) {
session.webrtc_connected = connected session.webrtcConnected = connected
session.manager.emmiter.Emit("state_changed", session) session.manager.emmiter.Emit("state_changed", session)
if !connected { if !connected {
session.webrtc_peer = nil session.webrtcPeer = nil
} }
} }
func (session *SessionCtx) GetWebRTCPeer() types.WebRTCPeer { func (session *SessionCtx) GetWebRTCPeer() types.WebRTCPeer {
return session.webrtc_peer return session.webrtcPeer
} }

4
internal/types/session.go
View File

@ -49,13 +49,13 @@ type Session interface {
GetState() MemberState GetState() MemberState
// websocket // websocket
SetWebSocketPeer(websocket_peer WebSocketPeer) SetWebSocketPeer(websocketPeer WebSocketPeer)
SetWebSocketConnected(connected bool) SetWebSocketConnected(connected bool)
Send(v interface{}) error Send(v interface{}) error
Disconnect(reason string) error Disconnect(reason string) error
// webrtc // webrtc
SetWebRTCPeer(webrtc_peer WebRTCPeer) SetWebRTCPeer(webrtcPeer WebRTCPeer)
SetWebRTCConnected(connected bool) SetWebRTCConnected(connected bool)
GetWebRTCPeer() WebRTCPeer GetWebRTCPeer() WebRTCPeer
} }

18
internal/webrtc/peer.go
View File

@ -9,25 +9,25 @@ type WebRTCPeerCtx struct {
changeVideo func(videoID string) error changeVideo func(videoID string) error
} }
func (webrtc_peer *WebRTCPeerCtx) SignalAnswer(sdp string) error { func (peer *WebRTCPeerCtx) SignalAnswer(sdp string) error {
return webrtc_peer.connection.SetRemoteDescription(webrtc.SessionDescription{ return peer.connection.SetRemoteDescription(webrtc.SessionDescription{
SDP: sdp, SDP: sdp,
Type: webrtc.SDPTypeAnswer, Type: webrtc.SDPTypeAnswer,
}) })
} }
func (webrtc_peer *WebRTCPeerCtx) SignalCandidate(candidate webrtc.ICECandidateInit) error { func (peer *WebRTCPeerCtx) SignalCandidate(candidate webrtc.ICECandidateInit) error {
return webrtc_peer.connection.AddICECandidate(candidate) return peer.connection.AddICECandidate(candidate)
} }
func (webrtc_peer *WebRTCPeerCtx) SetVideoID(videoID string) error { func (peer *WebRTCPeerCtx) SetVideoID(videoID string) error {
return webrtc_peer.changeVideo(videoID) return peer.changeVideo(videoID)
} }
func (webrtc_peer *WebRTCPeerCtx) Destroy() error { func (peer *WebRTCPeerCtx) Destroy() error {
if webrtc_peer.connection == nil || webrtc_peer.connection.ConnectionState() != webrtc.PeerConnectionStateConnected { if peer.connection == nil || peer.connection.ConnectionState() != webrtc.PeerConnectionStateConnected {
return nil return nil
} }
return webrtc_peer.connection.Close() return peer.connection.Close()
} }

12
internal/webrtc/send.go
View File

@ -28,8 +28,8 @@ type PayloadCursorImage struct {
Yhot uint16 Yhot uint16
} }
func (webrtc_peer *WebRTCPeerCtx) SendCursorPosition(x, y int) error { func (peer *WebRTCPeerCtx) SendCursorPosition(x, y int) error {
if webrtc_peer.dataChannel == nil { if peer.dataChannel == nil {
return fmt.Errorf("no data channel") return fmt.Errorf("no data channel")
} }
@ -47,11 +47,11 @@ func (webrtc_peer *WebRTCPeerCtx) SendCursorPosition(x, y int) error {
return err return err
} }
return webrtc_peer.dataChannel.Send(buffer.Bytes()) return peer.dataChannel.Send(buffer.Bytes())
} }
func (webrtc_peer *WebRTCPeerCtx) SendCursorImage(cur *types.CursorImage) error { func (peer *WebRTCPeerCtx) SendCursorImage(cur *types.CursorImage) error {
if webrtc_peer.dataChannel == nil { if peer.dataChannel == nil {
return fmt.Errorf("no data channel") return fmt.Errorf("no data channel")
} }
@ -81,5 +81,5 @@ func (webrtc_peer *WebRTCPeerCtx) SendCursorImage(cur *types.CursorImage) error
return err return err
} }
return webrtc_peer.dataChannel.Send(buffer.Bytes()) return peer.dataChannel.Send(buffer.Bytes())
} }

28
internal/websocket/filechooserdialog.go
View File

@ -6,41 +6,41 @@ import (
"demodesk/neko/internal/types/message" "demodesk/neko/internal/types/message"
) )
func (ws *WebSocketManagerCtx) fileChooserDialogEvents() { func (manager *WebSocketManagerCtx) fileChooserDialogEvents() {
var file_chooser_dialog_member types.Session var file_chooser_dialog_member types.Session
// when dialog opens, everyone should be notified. // when dialog opens, everyone should be notified.
ws.desktop.OnFileChooserDialogOpened(func() { manager.desktop.OnFileChooserDialogOpened(func() {
ws.logger.Info().Msg("FileChooserDialog opened") manager.logger.Info().Msg("FileChooserDialog opened")
host := ws.sessions.GetHost() host := manager.sessions.GetHost()
if host == nil { if host == nil {
ws.logger.Warn().Msg("no host for FileChooserDialog found, closing") manager.logger.Warn().Msg("no host for FileChooserDialog found, closing")
go ws.desktop.CloseFileChooserDialog() go manager.desktop.CloseFileChooserDialog()
return return
} }
file_chooser_dialog_member = host file_chooser_dialog_member = host
go ws.sessions.Broadcast(message.MemberID{ go manager.sessions.Broadcast(message.MemberID{
Event: event.FILE_CHOOSER_DIALOG_OPENED, Event: event.FILE_CHOOSER_DIALOG_OPENED,
ID: host.ID(), ID: host.ID(),
}, nil) }, nil)
}) })
// when dialog closes, everyone should be notified. // when dialog closes, everyone should be notified.
ws.desktop.OnFileChooserDialogClosed(func() { manager.desktop.OnFileChooserDialogClosed(func() {
ws.logger.Info().Msg("FileChooserDialog closed") manager.logger.Info().Msg("FileChooserDialog closed")
file_chooser_dialog_member = nil file_chooser_dialog_member = nil
go ws.sessions.Broadcast(message.MemberID{ go manager.sessions.Broadcast(message.MemberID{
Event: event.FILE_CHOOSER_DIALOG_CLOSED, Event: event.FILE_CHOOSER_DIALOG_CLOSED,
}, nil) }, nil)
}) })
// when new user joins, and someone holds dialog, he shouldd be notified about it. // when new user joins, and someone holds dialog, he shouldd be notified about it.
ws.sessions.OnConnected(func(session types.Session) { manager.sessions.OnConnected(func(session types.Session) {
if file_chooser_dialog_member == nil { if file_chooser_dialog_member == nil {
return return
} }
@ -49,7 +49,7 @@ func (ws *WebSocketManagerCtx) fileChooserDialogEvents() {
Event: event.FILE_CHOOSER_DIALOG_OPENED, Event: event.FILE_CHOOSER_DIALOG_OPENED,
ID: file_chooser_dialog_member.ID(), ID: file_chooser_dialog_member.ID(),
}); err != nil { }); err != nil {
ws.logger.Warn(). manager.logger.Warn().
Str("id", session.ID()). Str("id", session.ID()).
Err(err). Err(err).
Msgf("could not send event `%s` to session", event.FILE_CHOOSER_DIALOG_OPENED) Msgf("could not send event `%s` to session", event.FILE_CHOOSER_DIALOG_OPENED)
@ -57,7 +57,7 @@ func (ws *WebSocketManagerCtx) fileChooserDialogEvents() {
}) })
// when user, that holds dialog, disconnects, it should be closed. // when user, that holds dialog, disconnects, it should be closed.
ws.sessions.OnDisconnected(func(session types.Session) { manager.sessions.OnDisconnected(func(session types.Session) {
if file_chooser_dialog_member == nil { if file_chooser_dialog_member == nil {
return return
} }
@ -66,6 +66,6 @@ func (ws *WebSocketManagerCtx) fileChooserDialogEvents() {
return return
} }
ws.desktop.CloseFileChooserDialog() manager.desktop.CloseFileChooserDialog()
}) })
} }

110
internal/websocket/manager.go
View File

@ -43,64 +43,64 @@ type WebSocketManagerCtx struct {
shutdown chan bool shutdown chan bool
} }
func (ws *WebSocketManagerCtx) Start() { func (manager *WebSocketManagerCtx) Start() {
ws.sessions.OnCreated(func(session types.Session) { manager.sessions.OnCreated(func(session types.Session) {
if err := ws.handler.SessionCreated(session); err != nil { if err := manager.handler.SessionCreated(session); err != nil {
ws.logger.Warn().Str("id", session.ID()).Err(err).Msg("session created with an error") manager.logger.Warn().Str("id", session.ID()).Err(err).Msg("session created with an error")
} else { } else {
ws.logger.Debug().Str("id", session.ID()).Msg("session created") manager.logger.Debug().Str("id", session.ID()).Msg("session created")
} }
}) })
ws.sessions.OnDeleted(func(session types.Session) { manager.sessions.OnDeleted(func(session types.Session) {
if err := ws.handler.SessionDeleted(session); err != nil { if err := manager.handler.SessionDeleted(session); err != nil {
ws.logger.Warn().Str("id", session.ID()).Err(err).Msg("session deleted with an error") manager.logger.Warn().Str("id", session.ID()).Err(err).Msg("session deleted with an error")
} else { } else {
ws.logger.Debug().Str("id", session.ID()).Msg("session deleted") manager.logger.Debug().Str("id", session.ID()).Msg("session deleted")
} }
}) })
ws.sessions.OnConnected(func(session types.Session) { manager.sessions.OnConnected(func(session types.Session) {
if err := ws.handler.SessionConnected(session); err != nil { if err := manager.handler.SessionConnected(session); err != nil {
ws.logger.Warn().Str("id", session.ID()).Err(err).Msg("session connected with an error") manager.logger.Warn().Str("id", session.ID()).Err(err).Msg("session connected with an error")
} else { } else {
ws.logger.Debug().Str("id", session.ID()).Msg("session connected") manager.logger.Debug().Str("id", session.ID()).Msg("session connected")
} }
}) })
ws.sessions.OnDisconnected(func(session types.Session) { manager.sessions.OnDisconnected(func(session types.Session) {
if err := ws.handler.SessionDisconnected(session); err != nil { if err := manager.handler.SessionDisconnected(session); err != nil {
ws.logger.Warn().Str("id", session.ID()).Err(err).Msg("session disconnected with an error") manager.logger.Warn().Str("id", session.ID()).Err(err).Msg("session disconnected with an error")
} else { } else {
ws.logger.Debug().Str("id", session.ID()).Msg("session disconnected") manager.logger.Debug().Str("id", session.ID()).Msg("session disconnected")
} }
}) })
ws.sessions.OnProfileChanged(func(session types.Session) { manager.sessions.OnProfileChanged(func(session types.Session) {
if err := ws.handler.SessionProfileChanged(session); err != nil { if err := manager.handler.SessionProfileChanged(session); err != nil {
ws.logger.Warn().Str("id", session.ID()).Err(err).Msg("session profile changed with an error") manager.logger.Warn().Str("id", session.ID()).Err(err).Msg("session profile changed with an error")
} else { } else {
ws.logger.Debug().Str("id", session.ID()).Msg("session profile changed") manager.logger.Debug().Str("id", session.ID()).Msg("session profile changed")
} }
}) })
ws.sessions.OnStateChanged(func(session types.Session) { manager.sessions.OnStateChanged(func(session types.Session) {
if err := ws.handler.SessionStateChanged(session); err != nil { if err := manager.handler.SessionStateChanged(session); err != nil {
ws.logger.Warn().Str("id", session.ID()).Err(err).Msg("session state changed with an error") manager.logger.Warn().Str("id", session.ID()).Err(err).Msg("session state changed with an error")
} else { } else {
ws.logger.Debug().Str("id", session.ID()).Msg("session state changed") manager.logger.Debug().Str("id", session.ID()).Msg("session state changed")
} }
}) })
ws.desktop.OnClipboardUpdated(func() { manager.desktop.OnClipboardUpdated(func() {
session := ws.sessions.GetHost() session := manager.sessions.GetHost()
if session == nil || !session.CanAccessClipboard() { if session == nil || !session.CanAccessClipboard() {
return return
} }
data, err := ws.desktop.ClipboardGetText() data, err := manager.desktop.ClipboardGetText()
if err != nil { if err != nil {
ws.logger.Warn().Err(err).Msg("could not get clipboard content") manager.logger.Warn().Err(err).Msg("could not get clipboard content")
return return
} }
@ -109,24 +109,24 @@ func (ws *WebSocketManagerCtx) Start() {
Text: data.Text, Text: data.Text,
// TODO: Send HTML? // TODO: Send HTML?
}); err != nil { }); err != nil {
ws.logger.Warn().Err(err).Msg("could not sync clipboard") manager.logger.Warn().Err(err).Msg("could not sync clipboard")
} }
}) })
ws.fileChooserDialogEvents() manager.fileChooserDialogEvents()
} }
func (ws *WebSocketManagerCtx) Shutdown() error { func (manager *WebSocketManagerCtx) Shutdown() error {
ws.shutdown <- true manager.shutdown <- true
return nil return nil
} }
func (ws *WebSocketManagerCtx) AddHandler(handler types.HandlerFunction) { func (manager *WebSocketManagerCtx) AddHandler(handler types.HandlerFunction) {
ws.handlers = append(ws.handlers, handler) manager.handlers = append(manager.handlers, handler)
} }
func (ws *WebSocketManagerCtx) Upgrade(w http.ResponseWriter, r *http.Request, checkOrigin types.CheckOrigin) error { func (manager *WebSocketManagerCtx) Upgrade(w http.ResponseWriter, r *http.Request, checkOrigin types.CheckOrigin) error {
ws.logger.Debug().Msg("attempting to upgrade connection") manager.logger.Debug().Msg("attempting to upgrade connection")
upgrader := websocket.Upgrader{ upgrader := websocket.Upgrader{
CheckOrigin: checkOrigin, CheckOrigin: checkOrigin,
@ -134,13 +134,13 @@ func (ws *WebSocketManagerCtx) Upgrade(w http.ResponseWriter, r *http.Request, c
connection, err := upgrader.Upgrade(w, r, nil) connection, err := upgrader.Upgrade(w, r, nil)
if err != nil { if err != nil {
ws.logger.Error().Err(err).Msg("failed to upgrade connection") manager.logger.Error().Err(err).Msg("failed to upgrade connection")
return err return err
} }
session, err := ws.sessions.AuthenticateRequest(r) session, err := manager.sessions.AuthenticateRequest(r)
if err != nil { if err != nil {
ws.logger.Warn().Err(err).Msg("authentication failed") manager.logger.Warn().Err(err).Msg("authentication failed")
// TODO: Refactor, return error code. // TODO: Refactor, return error code.
if err = connection.WriteJSON( if err = connection.WriteJSON(
@ -148,7 +148,7 @@ func (ws *WebSocketManagerCtx) Upgrade(w http.ResponseWriter, r *http.Request, c
Event: event.SYSTEM_DISCONNECT, Event: event.SYSTEM_DISCONNECT,
Message: err.Error(), Message: err.Error(),
}); err != nil { }); err != nil {
ws.logger.Error().Err(err).Msg("failed to send disconnect") manager.logger.Error().Err(err).Msg("failed to send disconnect")
} }
return connection.Close() return connection.Close()
@ -161,7 +161,7 @@ func (ws *WebSocketManagerCtx) Upgrade(w http.ResponseWriter, r *http.Request, c
Event: event.SYSTEM_DISCONNECT, Event: event.SYSTEM_DISCONNECT,
Message: "connection disabled", Message: "connection disabled",
}); err != nil { }); err != nil {
ws.logger.Error().Err(err).Msg("failed to send disconnect") manager.logger.Error().Err(err).Msg("failed to send disconnect")
} }
return connection.Close() return connection.Close()
@ -174,7 +174,7 @@ func (ws *WebSocketManagerCtx) Upgrade(w http.ResponseWriter, r *http.Request, c
Event: event.SYSTEM_DISCONNECT, Event: event.SYSTEM_DISCONNECT,
Message: "already connected", Message: "already connected",
}); err != nil { }); err != nil {
ws.logger.Error().Err(err).Msg("failed to send disconnect") manager.logger.Error().Err(err).Msg("failed to send disconnect")
} }
return connection.Close() return connection.Close()
@ -182,11 +182,11 @@ func (ws *WebSocketManagerCtx) Upgrade(w http.ResponseWriter, r *http.Request, c
session.SetWebSocketPeer(&WebSocketPeerCtx{ session.SetWebSocketPeer(&WebSocketPeerCtx{
session: session, session: session,
ws: ws, manager: manager,
connection: connection, connection: connection,
}) })
ws.logger. manager.logger.
Debug(). Debug().
Str("session", session.ID()). Str("session", session.ID()).
Str("address", connection.RemoteAddr().String()). Str("address", connection.RemoteAddr().String()).
@ -195,7 +195,7 @@ func (ws *WebSocketManagerCtx) Upgrade(w http.ResponseWriter, r *http.Request, c
session.SetWebSocketConnected(true) session.SetWebSocketConnected(true)
defer func() { defer func() {
ws.logger. manager.logger.
Debug(). Debug().
Str("session", session.ID()). Str("session", session.ID()).
Str("address", connection.RemoteAddr().String()). Str("address", connection.RemoteAddr().String()).
@ -204,11 +204,11 @@ func (ws *WebSocketManagerCtx) Upgrade(w http.ResponseWriter, r *http.Request, c
session.SetWebSocketConnected(false) session.SetWebSocketConnected(false)
}() }()
ws.handle(connection, session) manager.handle(connection, session)
return nil return nil
} }
func (ws *WebSocketManagerCtx) handle(connection *websocket.Conn, session types.Session) { func (manager *WebSocketManagerCtx) handle(connection *websocket.Conn, session types.Session) {
bytes := make(chan []byte) bytes := make(chan []byte)
cancel := make(chan struct{}) cancel := make(chan struct{})
@ -220,9 +220,9 @@ func (ws *WebSocketManagerCtx) handle(connection *websocket.Conn, session types.
_, raw, err := connection.ReadMessage() _, raw, err := connection.ReadMessage()
if err != nil { if err != nil {
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) { if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
ws.logger.Warn().Err(err).Msg("read message error") manager.logger.Warn().Err(err).Msg("read message error")
} else { } else {
ws.logger.Debug().Err(err).Msg("read message error") manager.logger.Debug().Err(err).Msg("read message error")
} }
close(cancel) close(cancel)
@ -236,14 +236,14 @@ func (ws *WebSocketManagerCtx) handle(connection *websocket.Conn, session types.
for { for {
select { select {
case raw := <-bytes: case raw := <-bytes:
ws.logger.Debug(). manager.logger.Debug().
Str("session", session.ID()). Str("session", session.ID()).
Str("address", connection.RemoteAddr().String()). Str("address", connection.RemoteAddr().String()).
Str("raw", string(raw)). Str("raw", string(raw)).
Msg("received message from client") Msg("received message from client")
handled := ws.handler.Message(session, raw) handled := manager.handler.Message(session, raw)
for _, handler := range ws.handlers { for _, handler := range manager.handlers {
if handled { if handled {
break break
} }
@ -252,13 +252,13 @@ func (ws *WebSocketManagerCtx) handle(connection *websocket.Conn, session types.
} }
if !handled { if !handled {
ws.logger.Warn().Msg("unhandled message") manager.logger.Warn().Msg("unhandled message")
} }
case <-cancel: case <-cancel:
return return
case <-ticker.C: case <-ticker.C:
if err := connection.WriteMessage(websocket.PingMessage, nil); err != nil { if err := connection.WriteMessage(websocket.PingMessage, nil); err != nil {
ws.logger.Error().Err(err).Msg("ping message has failed") manager.logger.Error().Err(err).Msg("ping message has failed")
return return
} }
} }

28
internal/websocket/peer.go
View File

@ -10,17 +10,17 @@ import (
) )
type WebSocketPeerCtx struct { type WebSocketPeerCtx struct {
session types.Session
ws *WebSocketManagerCtx
connection *websocket.Conn
mu sync.Mutex mu sync.Mutex
session types.Session
manager *WebSocketManagerCtx
connection *websocket.Conn
} }
func (websocket_peer *WebSocketPeerCtx) Send(v interface{}) error { func (peer *WebSocketPeerCtx) Send(v interface{}) error {
websocket_peer.mu.Lock() peer.mu.Lock()
defer websocket_peer.mu.Unlock() defer peer.mu.Unlock()
if websocket_peer.connection == nil { if peer.connection == nil {
return nil return nil
} }
@ -29,19 +29,19 @@ func (websocket_peer *WebSocketPeerCtx) Send(v interface{}) error {
return err return err
} }
websocket_peer.ws.logger.Debug(). peer.manager.logger.Debug().
Str("session", websocket_peer.session.ID()). Str("session", peer.session.ID()).
Str("address", websocket_peer.connection.RemoteAddr().String()). Str("address", peer.connection.RemoteAddr().String()).
Str("raw", string(raw)). Str("raw", string(raw)).
Msg("sending message to client") Msg("sending message to client")
return websocket_peer.connection.WriteMessage(websocket.TextMessage, raw) return peer.connection.WriteMessage(websocket.TextMessage, raw)
} }
func (websocket_peer *WebSocketPeerCtx) Destroy() error { func (peer *WebSocketPeerCtx) Destroy() error {
if websocket_peer.connection == nil { if peer.connection == nil {
return nil return nil
} }
return websocket_peer.connection.Close() return peer.connection.Close()
} }