1085

Committed 16 Dec 2020 02:51AM UTC coverage: 69.018%. First build

Build # 1085

Build Type

Pull #153

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/topfreegames/pitaya/commit/<a class=hub.com/topfreegames/pitaya/commit/<a class="double-link" href="https://git"><a class=hub.com/topfreegames/pitaya/commit/<a class="double-link" href="https://git"><a class=hub.com/topfreegames/pitaya/commit/<a class="double-link" href="https://git"><a class=hub.com/topfreegames/pitaya/commit/fbaa4264e99cd9a2df2fe8f8d486273db072997e">fbaa4264e">&lt;a href=&quot;https://github.com/topfreegames/pitaya/commit/</a><a class="double-link" href="https://github.com/topfreegames/pitaya/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/topfreegames/pitaya/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/topfreegames/pitaya/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/topfreegames/pitaya/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/topfreegames/pitaya/commit/fbaa4264e99cd9a2df2fe8f8d486273db072997e">fbaa4264e</a><a href="https://github.com/topfreegames/pitaya/commit/fbaa4264e99cd9a2df2fe8f8d486273db072997e">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/topfreegames/pitaya/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/topfreegames/pitaya/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/topfreegames/pitaya/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/topfreegames/pitaya/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/topfreegames/pitaya/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/topfreegames/pitaya/commit/fbaa4264e99cd9a2df2fe8f8d486273db072997e&quot;&gt;fbaa4264e&lt;/a&gt;&lt;a href=&quot;https://github.com/topfreegames/pitaya/commit/fbaa4264e99cd9a2df2fe8f8d486273db072997e&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/topfreegames/pitaya/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/topfree... (continued)

Pull Request Pull Request #153: BeforeHandler return context

Run Details

11 of 11 new or added lines in 3 files covered. (100.0%)

4589 of 6649 relevant lines covered (69.02%)

0.77 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.47

/service/handler.go

// Copyright (c) nano Author and TFG Co. All Rights Reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.

package service

import (
        "context"
        "encoding/json"
        "fmt"
        "strings"
        "time"

        "github.com/topfreegames/pitaya/acceptor"

        "github.com/google/uuid"
        opentracing "github.com/opentracing/opentracing-go"
        "github.com/topfreegames/pitaya/agent"
        "github.com/topfreegames/pitaya/cluster"
        "github.com/topfreegames/pitaya/component"
        "github.com/topfreegames/pitaya/conn/codec"
        "github.com/topfreegames/pitaya/conn/message"
        "github.com/topfreegames/pitaya/conn/packet"
        "github.com/topfreegames/pitaya/constants"
        pcontext "github.com/topfreegames/pitaya/context"
        "github.com/topfreegames/pitaya/docgenerator"
        e "github.com/topfreegames/pitaya/errors"
        "github.com/topfreegames/pitaya/logger"
        "github.com/topfreegames/pitaya/metrics"
        "github.com/topfreegames/pitaya/route"
        "github.com/topfreegames/pitaya/serialize"
        "github.com/topfreegames/pitaya/session"
        "github.com/topfreegames/pitaya/timer"
        "github.com/topfreegames/pitaya/tracing"
)

var (
        handlers    = make(map[string]*component.Handler) // all handler method
        handlerType = "handler"
)

type (
        // HandlerService service
        HandlerService struct {
                appDieChan         chan bool             // die channel app
                chLocalProcess     chan unhandledMessage // channel of messages that will be processed locally
                chRemoteProcess    chan unhandledMessage // channel of messages that will be processed remotely
                decoder            codec.PacketDecoder   // binary decoder
                encoder            codec.PacketEncoder   // binary encoder
                heartbeatTimeout   time.Duration
                messagesBufferSize int
                remoteService      *RemoteService
                serializer         serialize.Serializer          // message serializer
                server             *cluster.Server               // server obj
                services           map[string]*component.Service // all registered service
                messageEncoder     message.Encoder
                metricsReporters   []metrics.Reporter
        }

        unhandledMessage struct {
                ctx   context.Context
                agent *agent.Agent
                route *route.Route
                msg   *message.Message
        }
)

// NewHandlerService creates and returns a new handler service
func NewHandlerService(
        dieChan chan bool,
        packetDecoder codec.PacketDecoder,
        packetEncoder codec.PacketEncoder,
        serializer serialize.Serializer,
        heartbeatTime time.Duration,
        messagesBufferSize,
        localProcessBufferSize,
        remoteProcessBufferSize int,
        server *cluster.Server,
        remoteService *RemoteService,
        messageEncoder message.Encoder,
        metricsReporters []metrics.Reporter,
) *HandlerService {
        h := &HandlerService{
                services:           make(map[string]*component.Service),
                chLocalProcess:     make(chan unhandledMessage, localProcessBufferSize),
                chRemoteProcess:    make(chan unhandledMessage, remoteProcessBufferSize),
                decoder:            packetDecoder,
                encoder:            packetEncoder,
                messagesBufferSize: messagesBufferSize,
                serializer:         serializer,
                heartbeatTimeout:   heartbeatTime,
                appDieChan:         dieChan,
                server:             server,
                remoteService:      remoteService,
                messageEncoder:     messageEncoder,
                metricsReporters:   metricsReporters,
        }

        return h
}

// Dispatch message to corresponding logic handler
func (h *HandlerService) Dispatch(thread int) {
        // TODO: This timer is being stopped multiple times, it probably doesn't need to be stopped here
        defer timer.GlobalTicker.Stop()

        for {
                // Calls to remote servers block calls to local server
                select {
                case lm := <-h.chLocalProcess:
                        metrics.ReportMessageProcessDelayFromCtx(lm.ctx, h.metricsReporters, "local")
                        h.localProcess(lm.ctx, lm.agent, lm.route, lm.msg)

                case rm := <-h.chRemoteProcess:
                        metrics.ReportMessageProcessDelayFromCtx(rm.ctx, h.metricsReporters, "remote")
                        h.remoteService.remoteProcess(rm.ctx, nil, rm.agent, rm.route, rm.msg)

                case <-timer.GlobalTicker.C: // execute cron task
                        timer.Cron()

                case t := <-timer.Manager.ChCreatedTimer: // new Timers
                        timer.AddTimer(t)

                case id := <-timer.Manager.ChClosingTimer: // closing Timers
                        timer.RemoveTimer(id)
                }
        }
}

// Register registers components
func (h *HandlerService) Register(comp component.Component, opts []component.Option) error {
        s := component.NewService(comp, opts)

        if _, ok := h.services[s.Name]; ok {
                return fmt.Errorf("handler: service already defined: %s", s.Name)
        }

        if err := s.ExtractHandler(); err != nil {
                return err
        }

        // register all handlers
        h.services[s.Name] = s
        for name, handler := range s.Handlers {
                handlers[fmt.Sprintf("%s.%s", s.Name, name)] = handler
        }
        return nil
}

// Handle handles messages from a conn
func (h *HandlerService) Handle(conn acceptor.PlayerConn) {
        // create a client agent and startup write goroutine
        a := agent.NewAgent(conn, h.decoder, h.encoder, h.serializer, h.heartbeatTimeout, h.messagesBufferSize, h.appDieChan, h.messageEncoder, h.metricsReporters)

        // startup agent goroutine
        go a.Handle()

        logger.Log.Debugf("New session established: %s", a.String())

        // guarantee agent related resource is destroyed
        defer func() {
                a.Session.Close()
                logger.Log.Debugf("Session read goroutine exit, SessionID=%d, UID=%s", a.Session.ID(), a.Session.UID())
        }()

        for {
                msg, err := conn.GetNextMessage()

                if err != nil {
                        if err != constants.ErrConnectionClosed {
                                logger.Log.Errorf("Error reading next available message: %s", err.Error())
                        }

                        return
                }

                packets, err := h.decoder.Decode(msg)
                if err != nil {
                        logger.Log.Errorf("Failed to decode message: %s", err.Error())
                        return
                }

                if len(packets) < 1 {
                        logger.Log.Warnf("Read no packets, data: %v", msg)
                        continue
                }

                // process all packet
                for i := range packets {
                        if err := h.processPacket(a, packets[i]); err != nil {
                                logger.Log.Errorf("Failed to process packet: %s", err.Error())
                                return
                        }
                }
        }
}

func (h *HandlerService) processPacket(a *agent.Agent, p *packet.Packet) error {
        switch p.Type {
        case packet.Handshake:
                logger.Log.Debug("Received handshake packet")
                if err := a.SendHandshakeResponse(); err != nil {
                        logger.Log.Errorf("Error sending handshake response: %s", err.Error())
                        return err
                }
                logger.Log.Debugf("Session handshake Id=%d, Remote=%s", a.Session.ID(), a.RemoteAddr())

                // Parse the json sent with the handshake by the client
                handshakeData := &session.HandshakeData{}
                err := json.Unmarshal(p.Data, handshakeData)
                if err != nil {
                        a.SetStatus(constants.StatusClosed)
                        return fmt.Errorf("Invalid handshake data. Id=%d", a.Session.ID())
                }

                a.Session.SetHandshakeData(handshakeData)
                a.SetStatus(constants.StatusHandshake)
                err = a.Session.Set(constants.IPVersionKey, a.IPVersion())
                if err != nil {
                        logger.Log.Warnf("failed to save ip version on session: %q\n", err)
                }

                logger.Log.Debug("Successfully saved handshake data")

        case packet.HandshakeAck:
                a.SetStatus(constants.StatusWorking)
                logger.Log.Debugf("Receive handshake ACK Id=%d, Remote=%s", a.Session.ID(), a.RemoteAddr())

        case packet.Data:
                if a.GetStatus() < constants.StatusWorking {
                        return fmt.Errorf("receive data on socket which is not yet ACK, session will be closed immediately, remote=%s",
                                a.RemoteAddr().String())
                }

                msg, err := message.Decode(p.Data)
                if err != nil {
                        return err
                }
                h.processMessage(a, msg)

        case packet.Heartbeat:
                // expected
        }

        a.SetLastAt()
        return nil
}

func (h *HandlerService) processMessage(a *agent.Agent, msg *message.Message) {
        requestID := uuid.New()
        ctx := pcontext.AddToPropagateCtx(context.Background(), constants.StartTimeKey, time.Now().UnixNano())
        ctx = pcontext.AddToPropagateCtx(ctx, constants.RouteKey, msg.Route)
        ctx = pcontext.AddToPropagateCtx(ctx, constants.RequestIDKey, requestID.String())
        tags := opentracing.Tags{
                "local.id":   h.server.ID,
                "span.kind":  "server",
                "msg.type":   strings.ToLower(msg.Type.String()),
                "user.id":    a.Session.UID(),
                "request.id": requestID.String(),
        }
        ctx = tracing.StartSpan(ctx, msg.Route, tags)
        ctx = context.WithValue(ctx, constants.SessionCtxKey, a.Session)

        r, err := route.Decode(msg.Route)
        if err != nil {
                logger.Log.Errorf("Failed to decode route: %s", err.Error())
                a.AnswerWithError(ctx, msg.ID, e.NewError(err, e.ErrBadRequestCode))
                return
        }

        if r.SvType == "" {
                r.SvType = h.server.Type
        }

        message := unhandledMessage{
                ctx:   ctx,
                agent: a,
                route: r,
                msg:   msg,
        }
        if r.SvType == h.server.Type {
                h.chLocalProcess <- message
        } else {
                if h.remoteService != nil {
                        h.chRemoteProcess <- message
                } else {
                        logger.Log.Warnf("request made to another server type but no remoteService running")
                }
        }
}

func (h *HandlerService) localProcess(ctx context.Context, a *agent.Agent, route *route.Route, msg *message.Message) {
        var mid uint
        switch msg.Type {
        case message.Request:
                mid = msg.ID
        case message.Notify:
                mid = 0
        }

        ret, err := processHandlerMessage(ctx, route, h.serializer, a.Session, msg.Data, msg.Type, false)
        if msg.Type != message.Notify {
                if err != nil {
                        logger.Log.Errorf("Failed to process handler message: %s", err.Error())
                        a.AnswerWithError(ctx, mid, err)
                } else {
                        err := a.Session.ResponseMID(ctx, mid, ret)
                        if err != nil {
                                tracing.FinishSpan(ctx, err)
                                metrics.ReportTimingFromCtx(ctx, h.metricsReporters, handlerType, err)
                        }
                }
        } else {
                metrics.ReportTimingFromCtx(ctx, h.metricsReporters, handlerType, nil)
                tracing.FinishSpan(ctx, err)
        }
}

// DumpServices outputs all registered services
func (h *HandlerService) DumpServices() {
        for name := range handlers {
                logger.Log.Infof("registered handler %s, isRawArg: %v", name, handlers[name].IsRawArg)
        }
}

// Docs returns documentation for handlers
func (h *HandlerService) Docs(getPtrNames bool) (map[string]interface{}, error) {
        if h == nil {
                return map[string]interface{}{}, nil
        }
        return docgenerator.HandlersDocs(h.server.Type, h.services, getPtrNames)
}

1	// Copyright (c) nano Author and TFG Co. All Rights Reserved.
2	//
3	// Permission is hereby granted, free of charge, to any person obtaining a copy
4	// of this software and associated documentation files (the "Software"), to deal
5	// in the Software without restriction, including without limitation the rights
6	// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
7	// copies of the Software, and to permit persons to whom the Software is
8	// furnished to do so, subject to the following conditions:
9	//
10	// The above copyright notice and this permission notice shall be included in all
11	// copies or substantial portions of the Software.
12	//
13	// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
14	// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15	// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16	// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17	// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
18	// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
19	// SOFTWARE.
20
21	package service
22
23	import (
24	"context"
25	"encoding/json"
26	"fmt"
27	"strings"
28	"time"
29
30	"github.com/topfreegames/pitaya/acceptor"
31
32	"github.com/google/uuid"
33	opentracing "github.com/opentracing/opentracing-go"
34	"github.com/topfreegames/pitaya/agent"
35	"github.com/topfreegames/pitaya/cluster"
36	"github.com/topfreegames/pitaya/component"
37	"github.com/topfreegames/pitaya/conn/codec"
38	"github.com/topfreegames/pitaya/conn/message"
39	"github.com/topfreegames/pitaya/conn/packet"
40	"github.com/topfreegames/pitaya/constants"
41	pcontext "github.com/topfreegames/pitaya/context"
42	"github.com/topfreegames/pitaya/docgenerator"
43	e "github.com/topfreegames/pitaya/errors"
44	"github.com/topfreegames/pitaya/logger"
45	"github.com/topfreegames/pitaya/metrics"
46	"github.com/topfreegames/pitaya/route"
47	"github.com/topfreegames/pitaya/serialize"
48	"github.com/topfreegames/pitaya/session"
49	"github.com/topfreegames/pitaya/timer"
50	"github.com/topfreegames/pitaya/tracing"
51	)
52
53	var (
54	handlers = make(map[string]*component.Handler) // all handler method
55	handlerType = "handler"
56	)
57
58	type (
59	// HandlerService service
60	HandlerService struct {
61	appDieChan chan bool // die channel app
62	chLocalProcess chan unhandledMessage // channel of messages that will be processed locally
63	chRemoteProcess chan unhandledMessage // channel of messages that will be processed remotely
64	decoder codec.PacketDecoder // binary decoder
65	encoder codec.PacketEncoder // binary encoder
66	heartbeatTimeout time.Duration
67	messagesBufferSize int
68	remoteService *RemoteService
69	serializer serialize.Serializer // message serializer
70	server *cluster.Server // server obj
71	services map[string]*component.Service // all registered service
72	messageEncoder message.Encoder
73	metricsReporters []metrics.Reporter
74	}
75
76	unhandledMessage struct {
77	ctx context.Context
78	agent *agent.Agent
79	route *route.Route
80	msg *message.Message
81	}
82	)
83
84	// NewHandlerService creates and returns a new handler service
85	func NewHandlerService(
86	dieChan chan bool,
87	packetDecoder codec.PacketDecoder,
88	packetEncoder codec.PacketEncoder,
89	serializer serialize.Serializer,
90	heartbeatTime time.Duration,
91	messagesBufferSize,
92	localProcessBufferSize,
93	remoteProcessBufferSize int,
94	server *cluster.Server,
95	remoteService *RemoteService,
96	messageEncoder message.Encoder,
97	metricsReporters []metrics.Reporter,
98	) *HandlerService {	1✔
99	h := &HandlerService{	1✔
100	services: make(map[string]*component.Service),	1✔
101	chLocalProcess: make(chan unhandledMessage, localProcessBufferSize),	1✔
102	chRemoteProcess: make(chan unhandledMessage, remoteProcessBufferSize),	1✔
103	decoder: packetDecoder,	1✔
104	encoder: packetEncoder,	1✔
105	messagesBufferSize: messagesBufferSize,	1✔
106	serializer: serializer,	1✔
107	heartbeatTimeout: heartbeatTime,	1✔
108	appDieChan: dieChan,	1✔
109	server: server,	1✔
110	remoteService: remoteService,	1✔
111	messageEncoder: messageEncoder,	1✔
112	metricsReporters: metricsReporters,	1✔
113	}	1✔
114		1✔
115	return h	1✔
116	}	1✔
117
118	// Dispatch message to corresponding logic handler
119	func (h *HandlerService) Dispatch(thread int) {	×
120	// TODO: This timer is being stopped multiple times, it probably doesn't need to be stopped here	×
121	defer timer.GlobalTicker.Stop()	×
122		×
123	for {	×
124	// Calls to remote servers block calls to local server	×
125	select {	×
126	case lm := <-h.chLocalProcess:	×
127	metrics.ReportMessageProcessDelayFromCtx(lm.ctx, h.metricsReporters, "local")	×
128	h.localProcess(lm.ctx, lm.agent, lm.route, lm.msg)	×
129
130	case rm := <-h.chRemoteProcess:	×
131	metrics.ReportMessageProcessDelayFromCtx(rm.ctx, h.metricsReporters, "remote")	×
132	h.remoteService.remoteProcess(rm.ctx, nil, rm.agent, rm.route, rm.msg)	×
133
134	case <-timer.GlobalTicker.C: // execute cron task	×
135	timer.Cron()	×
136
137	case t := <-timer.Manager.ChCreatedTimer: // new Timers	×
138	timer.AddTimer(t)	×
139
140	case id := <-timer.Manager.ChClosingTimer: // closing Timers	×
141	timer.RemoveTimer(id)	×
142	}
143	}
144	}
145
146	// Register registers components
147	func (h *HandlerService) Register(comp component.Component, opts []component.Option) error {	1✔
148	s := component.NewService(comp, opts)	1✔
149		1✔
150	if _, ok := h.services[s.Name]; ok {	2✔
151	return fmt.Errorf("handler: service already defined: %s", s.Name)	1✔
152	}	1✔
153
154	if err := s.ExtractHandler(); err != nil {	2✔
155	return err	1✔
156	}	1✔
157
158	// register all handlers
159	h.services[s.Name] = s	1✔
160	for name, handler := range s.Handlers {	2✔
161	handlers[fmt.Sprintf("%s.%s", s.Name, name)] = handler	1✔
162	}	1✔
163	return nil	1✔
164	}
165
166	// Handle handles messages from a conn
167	func (h *HandlerService) Handle(conn acceptor.PlayerConn) {	1✔
168	// create a client agent and startup write goroutine	1✔
169	a := agent.NewAgent(conn, h.decoder, h.encoder, h.serializer, h.heartbeatTimeout, h.messagesBufferSize, h.appDieChan, h.messageEncoder, h.metricsReporters)	1✔
170		1✔
171	// startup agent goroutine	1✔
172	go a.Handle()	1✔
173		1✔
174	logger.Log.Debugf("New session established: %s", a.String())	1✔
175		1✔
176	// guarantee agent related resource is destroyed	1✔
177	defer func() {	2✔
178	a.Session.Close()	1✔
179	logger.Log.Debugf("Session read goroutine exit, SessionID=%d, UID=%s", a.Session.ID(), a.Session.UID())	1✔
180	}()	1✔
181
182	for {	2✔
183	msg, err := conn.GetNextMessage()	1✔
184		1✔
185	if err != nil {	2✔
186	if err != constants.ErrConnectionClosed {	2✔
187	logger.Log.Errorf("Error reading next available message: %s", err.Error())	1✔
188	}	1✔
189
190	return	1✔
191	}
192
193	packets, err := h.decoder.Decode(msg)	1✔
194	if err != nil {	1✔
195	logger.Log.Errorf("Failed to decode message: %s", err.Error())	×
196	return	×
197	}	×
198
199	if len(packets) < 1 {	1✔
200	logger.Log.Warnf("Read no packets, data: %v", msg)	×
201	continue	×
202	}
203
204	// process all packet
205	for i := range packets {	2✔
206	if err := h.processPacket(a, packets[i]); err != nil {	1✔
207	logger.Log.Errorf("Failed to process packet: %s", err.Error())	×
208	return	×
209	}	×
210	}
211	}
212	}
213
214	func (h HandlerService) processPacket(a agent.Agent, p *packet.Packet) error {	1✔
215	switch p.Type {	1✔
216	case packet.Handshake:	1✔
217	logger.Log.Debug("Received handshake packet")	1✔
218	if err := a.SendHandshakeResponse(); err != nil {	1✔
219	logger.Log.Errorf("Error sending handshake response: %s", err.Error())	×
220	return err	×
221	}	×
222	logger.Log.Debugf("Session handshake Id=%d, Remote=%s", a.Session.ID(), a.RemoteAddr())	1✔
223		1✔
224	// Parse the json sent with the handshake by the client	1✔
225	handshakeData := &session.HandshakeData{}	1✔
226	err := json.Unmarshal(p.Data, handshakeData)	1✔
227	if err != nil {	2✔
228	a.SetStatus(constants.StatusClosed)	1✔
229	return fmt.Errorf("Invalid handshake data. Id=%d", a.Session.ID())	1✔
230	}	1✔
231
232	a.Session.SetHandshakeData(handshakeData)	1✔
233	a.SetStatus(constants.StatusHandshake)	1✔
234	err = a.Session.Set(constants.IPVersionKey, a.IPVersion())	1✔
235	if err != nil {	1✔
236	logger.Log.Warnf("failed to save ip version on session: %q\n", err)	×
237	}	×
238
239	logger.Log.Debug("Successfully saved handshake data")	1✔
240
241	case packet.HandshakeAck:	1✔
242	a.SetStatus(constants.StatusWorking)	1✔
243	logger.Log.Debugf("Receive handshake ACK Id=%d, Remote=%s", a.Session.ID(), a.RemoteAddr())	1✔
244
245	case packet.Data:	1✔
246	if a.GetStatus() < constants.StatusWorking {	2✔
247	return fmt.Errorf("receive data on socket which is not yet ACK, session will be closed immediately, remote=%s",	1✔
248	a.RemoteAddr().String())	1✔
249	}	1✔
250
251	msg, err := message.Decode(p.Data)	1✔
252	if err != nil {	2✔
253	return err	1✔
254	}	1✔
255	h.processMessage(a, msg)	1✔
256
257	case packet.Heartbeat:	1✔
258	// expected
259	}
260
261	a.SetLastAt()	1✔
262	return nil	1✔
263	}
264
265	func (h HandlerService) processMessage(a agent.Agent, msg *message.Message) {	1✔
266	requestID := uuid.New()	1✔
267	ctx := pcontext.AddToPropagateCtx(context.Background(), constants.StartTimeKey, time.Now().UnixNano())	1✔
268	ctx = pcontext.AddToPropagateCtx(ctx, constants.RouteKey, msg.Route)	1✔
269	ctx = pcontext.AddToPropagateCtx(ctx, constants.RequestIDKey, requestID.String())	1✔
270	tags := opentracing.Tags{	1✔
271	"local.id": h.server.ID,	1✔
272	"span.kind": "server",	1✔
273	"msg.type": strings.ToLower(msg.Type.String()),	1✔
274	"user.id": a.Session.UID(),	1✔
275	"request.id": requestID.String(),	1✔
276	}	1✔
277	ctx = tracing.StartSpan(ctx, msg.Route, tags)	1✔
278	ctx = context.WithValue(ctx, constants.SessionCtxKey, a.Session)	1✔
279		1✔
280	r, err := route.Decode(msg.Route)	1✔
281	if err != nil {	2✔
282	logger.Log.Errorf("Failed to decode route: %s", err.Error())	1✔
283	a.AnswerWithError(ctx, msg.ID, e.NewError(err, e.ErrBadRequestCode))	1✔
284	return	1✔
285	}	1✔
286
287	if r.SvType == "" {	2✔
288	r.SvType = h.server.Type	1✔
289	}	1✔
290
291	message := unhandledMessage{	1✔
292	ctx: ctx,	1✔
293	agent: a,	1✔
294	route: r,	1✔
295	msg: msg,	1✔
296	}	1✔
297	if r.SvType == h.server.Type {	2✔
298	h.chLocalProcess <- message	1✔
299	} else {	2✔
300	if h.remoteService != nil {	2✔
301	h.chRemoteProcess <- message	1✔
302	} else {	1✔
303	logger.Log.Warnf("request made to another server type but no remoteService running")	×
304	}	×
305	}
306	}
307
308	func (h HandlerService) localProcess(ctx context.Context, a agent.Agent, route route.Route, msg message.Message) {	1✔
309	var mid uint	1✔
310	switch msg.Type {	1✔
311	case message.Request:	1✔
312	mid = msg.ID	1✔
313	case message.Notify:	×
314	mid = 0	×
315	}
316
317	ret, err := processHandlerMessage(ctx, route, h.serializer, a.Session, msg.Data, msg.Type, false)	1✔
318	if msg.Type != message.Notify {	2✔
319	if err != nil {	2✔
320	logger.Log.Errorf("Failed to process handler message: %s", err.Error())	1✔
321	a.AnswerWithError(ctx, mid, err)	1✔
322	} else {	2✔
323	err := a.Session.ResponseMID(ctx, mid, ret)	1✔
324	if err != nil {	1✔
325	tracing.FinishSpan(ctx, err)	×
326	metrics.ReportTimingFromCtx(ctx, h.metricsReporters, handlerType, err)	×
327	}	×
328	}
329	} else {	×
330	metrics.ReportTimingFromCtx(ctx, h.metricsReporters, handlerType, nil)	×
331	tracing.FinishSpan(ctx, err)	×
332	}	×
333	}
334
335	// DumpServices outputs all registered services
336	func (h *HandlerService) DumpServices() {	×
337	for name := range handlers {	×
338	logger.Log.Infof("registered handler %s, isRawArg: %v", name, handlers[name].IsRawArg)	×
339	}	×
340	}
341
342	// Docs returns documentation for handlers
343	func (h *HandlerService) Docs(getPtrNames bool) (map[string]interface{}, error) {	×
344	if h == nil {	×
345	return map[string]interface{}{}, nil	×
346	}	×
347	return docgenerator.HandlersDocs(h.server.Type, h.services, getPtrNames)	×
348	}

topfreegames / pitaya / 1085

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous