This commit is contained in:
crusader 2017-11-28 01:22:02 +09:00
parent 84676e199e
commit 6fbd3d2148
6 changed files with 28 additions and 12 deletions

View File

@ -3,6 +3,7 @@ package fasthttp
import ( import (
"fmt" "fmt"
"git.loafle.net/commons_go/rpc"
"git.loafle.net/commons_go/rpc/protocol" "git.loafle.net/commons_go/rpc/protocol"
"git.loafle.net/commons_go/websocket_fasthttp/websocket" "git.loafle.net/commons_go/websocket_fasthttp/websocket"
) )
@ -10,7 +11,7 @@ import (
type ServletHandlers struct { type ServletHandlers struct {
} }
func (sh *ServletHandlers) GetRequest(codec protocol.ServerCodec, reader interface{}) (protocol.ServerRequestCodec, error) { func (sh *ServletHandlers) GetRequest(servletCTX rpc.ServletContext, codec protocol.ServerCodec, reader interface{}) (protocol.ServerRequestCodec, error) {
conn := reader.(*websocket.Conn) conn := reader.(*websocket.Conn)
_, r, err := conn.NextReader() _, r, err := conn.NextReader()
@ -19,7 +20,7 @@ func (sh *ServletHandlers) GetRequest(codec protocol.ServerCodec, reader interfa
return requestCodec, err return requestCodec, err
} }
func (sh *ServletHandlers) SendResponse(requestCodec protocol.ServerRequestCodec, writer interface{}, result interface{}, err error) error { func (sh *ServletHandlers) SendResponse(servletCTX rpc.ServletContext, requestCodec protocol.ServerRequestCodec, writer interface{}, result interface{}, err error) error {
conn := writer.(*websocket.Conn) conn := writer.(*websocket.Conn)
wc, lerr := conn.NextWriter(websocket.TextMessage) wc, lerr := conn.NextWriter(websocket.TextMessage)
@ -40,7 +41,7 @@ func (sh *ServletHandlers) SendResponse(requestCodec protocol.ServerRequestCodec
return fmt.Errorf("Servlet Handler: SendResponse is not implemented") return fmt.Errorf("Servlet Handler: SendResponse is not implemented")
} }
func (sh *ServletHandlers) SendNotification(codec protocol.ServerCodec, writer interface{}, method string, args ...interface{}) error { func (sh *ServletHandlers) SendNotification(servletCTX rpc.ServletContext, codec protocol.ServerCodec, writer interface{}, method string, args ...interface{}) error {
conn := writer.(*websocket.Conn) conn := writer.(*websocket.Conn)
wc, lerr := conn.NextWriter(websocket.TextMessage) wc, lerr := conn.NextWriter(websocket.TextMessage)

View File

@ -1,5 +1,11 @@
package rpc package rpc
import cuc "git.loafle.net/commons_go/util/context"
const ( const (
DefaultPendingMessages = 32 * 1024 DefaultPendingMessages = 32 * 1024
) )
var (
ContentTypeKey = cuc.ContextKey("ContentType")
)

View File

@ -6,16 +6,17 @@ import (
"git.loafle.net/commons_go/logging" "git.loafle.net/commons_go/logging"
"git.loafle.net/commons_go/rpc/protocol" "git.loafle.net/commons_go/rpc/protocol"
cuc "git.loafle.net/commons_go/util/context"
) )
func NewServlet(servletCTX ServletContext, sh ServletHandler) Servlet { func NewServlet(sh ServletHandler) Servlet {
return &servlet{ return &servlet{
sh: sh, sh: sh,
} }
} }
type Servlet interface { type Servlet interface {
Start(contentType string, reader interface{}, writer interface{}) error Start(parentCTX cuc.Context, reader interface{}, writer interface{}) error
Stop() Stop()
Send(method string, args ...interface{}) (err error) Send(method string, args ...interface{}) (err error)
@ -28,7 +29,6 @@ type servlet struct {
sh ServletHandler sh ServletHandler
messageQueueChan chan *messageState messageQueueChan chan *messageState
contentType string
reader interface{} reader interface{}
writer interface{} writer interface{}
serverCodec protocol.ServerCodec serverCodec protocol.ServerCodec
@ -37,7 +37,7 @@ type servlet struct {
stopWg sync.WaitGroup stopWg sync.WaitGroup
} }
func (s *servlet) Start(contentType string, reader interface{}, writer interface{}) error { func (s *servlet) Start(parentCTX cuc.Context, reader interface{}, writer interface{}) error {
if nil == s.sh { if nil == s.sh {
panic("Servlet: servlet handler must be specified.") panic("Servlet: servlet handler must be specified.")
} }
@ -46,13 +46,13 @@ func (s *servlet) Start(contentType string, reader interface{}, writer interface
if s.stopChan != nil { if s.stopChan != nil {
return fmt.Errorf("Servlet: servlet is already running. Stop it before starting it again") return fmt.Errorf("Servlet: servlet is already running. Stop it before starting it again")
} }
servletCTX := s.sh.ServletContext(parentCTX)
sc, err := s.sh.getCodec(contentType) sc, err := s.sh.getCodec(servletCTX.GetAttribute(ContentTypeKey).(string))
if nil != err { if nil != err {
return err return err
} }
s.contentType = contentType
s.reader = reader s.reader = reader
s.writer = writer s.writer = writer
s.serverCodec = sc s.serverCodec = sc
@ -81,7 +81,6 @@ func (s *servlet) Stop() {
s.messageQueueChan = nil s.messageQueueChan = nil
s.contentType = ""
s.reader = nil s.reader = nil
s.writer = nil s.writer = nil
s.serverCodec = nil s.serverCodec = nil

View File

@ -12,7 +12,7 @@ type servletContext struct {
cuc.Context cuc.Context
} }
func NewServletContext(parent cuc.Context) ServletContext { func newServletContext(parent cuc.Context) ServletContext {
sCTX := &servletContext{} sCTX := &servletContext{}
sCTX.Context = cuc.NewContext(parent) sCTX.Context = cuc.NewContext(parent)

View File

@ -1,8 +1,13 @@
package rpc package rpc
import "git.loafle.net/commons_go/rpc/protocol" import (
"git.loafle.net/commons_go/rpc/protocol"
cuc "git.loafle.net/commons_go/util/context"
)
type ServletHandler interface { type ServletHandler interface {
ServletContext(parent cuc.Context) ServletContext
Init(servletCTX ServletContext) error Init(servletCTX ServletContext) error
GetRequest(servletCTX ServletContext, codec protocol.ServerCodec, reader interface{}) (protocol.ServerRequestCodec, error) GetRequest(servletCTX ServletContext, codec protocol.ServerCodec, reader interface{}) (protocol.ServerRequestCodec, error)

View File

@ -5,6 +5,7 @@ import (
"strings" "strings"
"git.loafle.net/commons_go/rpc/protocol" "git.loafle.net/commons_go/rpc/protocol"
cuc "git.loafle.net/commons_go/util/context"
) )
type ServletHandlers struct { type ServletHandlers struct {
@ -20,6 +21,10 @@ type ServletHandlers struct {
codecs map[string]protocol.ServerCodec codecs map[string]protocol.ServerCodec
} }
func (sh *ServletHandlers) ServletContext(parent cuc.Context) ServletContext {
return newServletContext(parent)
}
func (sh *ServletHandlers) Init(servletCTX ServletContext) error { func (sh *ServletHandlers) Init(servletCTX ServletContext) error {
return nil return nil
} }