diff --git a/apier/v1/accounts.go b/apier/v1/accounts.go index 16ce8fd74..fedf0b6b4 100644 --- a/apier/v1/accounts.go +++ b/apier/v1/accounts.go @@ -38,11 +38,6 @@ type AccountActionTiming struct { } func (apierSv1 *APIerSv1) GetAccountActionPlan(attrs *utils.TenantAccount, reply *[]*AccountActionTiming) error { - if err := config.ConReqs().VerifyAndGet(); err != nil { - return err - } - defer config.ConReqs().Putback() - if missing := utils.MissingStructFields(attrs, []string{"Tenant", "Account"}); len(missing) != 0 { return utils.NewErrMandatoryIeMissing(strings.Join(missing, ","), "") } @@ -91,10 +86,6 @@ type AttrRemoveActionTiming struct { // Removes an ActionTimings or parts of it depending on filters being set func (apierSv1 *APIerSv1) RemoveActionTiming(attrs *AttrRemoveActionTiming, reply *string) (err error) { - if err := config.ConReqs().VerifyAndGet(); err != nil { - return err - } - defer config.ConReqs().Putback() if missing := utils.MissingStructFields(attrs, []string{"ActionPlanId"}); len(missing) != 0 { // Only mandatory ActionPlanId return utils.NewErrMandatoryIeMissing(missing...) } @@ -182,10 +173,6 @@ func (apierSv1 *APIerSv1) RemoveActionTiming(attrs *AttrRemoveActionTiming, repl // Ads a new account into dataDb. If already defined, returns success. func (apierSv1 *APIerSv1) SetAccount(attr *utils.AttrSetAccount, reply *string) (err error) { - if err := config.ConReqs().VerifyAndGet(); err != nil { - return err - } - defer config.ConReqs().Putback() if missing := utils.MissingStructFields(attr, []string{"Tenant", "Account"}); len(missing) != 0 { return utils.NewErrMandatoryIeMissing(missing...) } @@ -307,10 +294,6 @@ func (apierSv1 *APIerSv1) SetAccount(attr *utils.AttrSetAccount, reply *string) } func (apierSv1 *APIerSv1) RemoveAccount(attr *utils.AttrRemoveAccount, reply *string) (err error) { - if err := config.ConReqs().VerifyAndGet(); err != nil { - return err - } - defer config.ConReqs().Putback() if missing := utils.MissingStructFields(attr, []string{"Tenant", "Account"}); len(missing) != 0 { return utils.NewErrMandatoryIeMissing(missing...) } @@ -369,10 +352,6 @@ func (apierSv1 *APIerSv1) RemoveAccount(attr *utils.AttrRemoveAccount, reply *st } func (apierSv1 *APIerSv1) GetAccounts(attr *utils.AttrGetAccounts, reply *[]interface{}) error { - if err := config.ConReqs().VerifyAndGet(); err != nil { - return err - } - defer config.ConReqs().Putback() if len(attr.Tenant) == 0 { return utils.NewErrMandatoryIeMissing("Tenant") } @@ -442,17 +421,9 @@ type AttrAddBalance struct { } func (apierSv1 *APIerSv1) AddBalance(attr *AttrAddBalance, reply *string) error { - if err := config.ConReqs().VerifyAndGet(); err != nil { - return err - } - defer config.ConReqs().Putback() return apierSv1.modifyBalance(utils.TOPUP, attr, reply) } func (apierSv1 *APIerSv1) DebitBalance(attr *AttrAddBalance, reply *string) error { - if err := config.ConReqs().VerifyAndGet(); err != nil { - return err - } - defer config.ConReqs().Putback() return apierSv1.modifyBalance(utils.DEBIT, attr, reply) } @@ -530,10 +501,6 @@ func (apierSv1 *APIerSv1) modifyBalance(aType string, attr *AttrAddBalance, repl // SetBalance sets the balance for the given account // if the account is not already created it will create the account also func (apierSv1 *APIerSv1) SetBalance(attr *utils.AttrSetBalance, reply *string) (err error) { - if err := config.ConReqs().VerifyAndGet(); err != nil { - return err - } - defer config.ConReqs().Putback() if missing := utils.MissingStructFields(attr, []string{"Tenant", "Account", "BalanceType"}); len(missing) != 0 { return utils.NewErrMandatoryIeMissing(missing...) } @@ -606,10 +573,6 @@ func (apierSv1 *APIerSv1) SetBalance(attr *utils.AttrSetBalance, reply *string) // RemoveBalances remove the matching balances for the account func (apierSv1 *APIerSv1) RemoveBalances(attr *utils.AttrSetBalance, reply *string) (err error) { - if err := config.ConReqs().VerifyAndGet(); err != nil { - return err - } - defer config.ConReqs().Putback() if missing := utils.MissingStructFields(attr, []string{"Tenant", "Account", "BalanceType"}); len(missing) != 0 { return utils.NewErrMandatoryIeMissing(missing...) } @@ -644,10 +607,6 @@ func (apierSv1 *APIerSv1) RemoveBalances(attr *utils.AttrSetBalance, reply *stri } func (apierSv1 *APIerSv1) GetAccountsCount(attr *utils.TenantArg, reply *int) (err error) { - if err := config.ConReqs().VerifyAndGet(); err != nil { - return err - } - defer config.ConReqs().Putback() if len(attr.Tenant) == 0 { return utils.NewErrMandatoryIeMissing("Tenant") } diff --git a/cmd/cgr-engine/cgr-engine.go b/cmd/cgr-engine/cgr-engine.go index 27ff175a0..81be70f59 100644 --- a/cmd/cgr-engine/cgr-engine.go +++ b/cmd/cgr-engine/cgr-engine.go @@ -236,11 +236,7 @@ func initLogger(cfg *config.CGRConfig) error { if *syslogger != "" { // Modify the log level if provided by command arguments sylogger = *syslogger } - err := utils.Newlogger(sylogger, cfg.GeneralCfg().NodeID) - if err != nil { - return err - } - return nil + return utils.Newlogger(sylogger, cfg.GeneralCfg().NodeID) } func initConfigSv1(internalConfigChan chan rpcclient.ClientConnector, @@ -394,7 +390,8 @@ func main() { lgLevel = *logLevel } utils.Logger.SetLogLevel(lgLevel) - + // init the concurrentRequests + utils.ConReqs = utils.NewConReqs(cfg.GeneralCfg().ConcurrentRequests, cfg.GeneralCfg().ConcurrentStrategy) utils.Logger.Info(fmt.Sprintf(" starting version <%s><%s>", vers, goVers)) cfg.LazySanityCheck() diff --git a/config/config.go b/config/config.go index a5e3105cd..cd9489e81 100755 --- a/config/config.go +++ b/config/config.go @@ -46,7 +46,6 @@ var ( dfltAstConnCfg *AsteriskConnCfg dfltLoaderConfig *LoaderSCfg dfltLoaderDataTypeConfig *LoaderDataType - conReqs *ConcReqs ) func newDbDefaults() dbDefaults { @@ -120,7 +119,6 @@ func (dbDflt dbDefaults) dbPass(dbType string, flagInput string) string { func init() { cgrCfg, _ = NewDefaultCGRConfig() dbDefaultsCfg = newDbDefaults() - conReqs = NewConReqs(cgrCfg.generalCfg.ConcurrentRequests, cgrCfg.generalCfg.ConcurrentStrategy) } // CgrConfig is used to retrieve system configuration from other packages @@ -128,11 +126,6 @@ func CgrConfig() *CGRConfig { return cgrCfg } -// ConReqs is used to retrieve system ConcurrentRequests counter from other packages -func ConReqs() *ConcReqs { - return conReqs -} - // SetCgrConfig is used to set system configuration from other places func SetCgrConfig(cfg *CGRConfig) { cgrCfg = cfg @@ -248,7 +241,6 @@ func NewCGRConfigFromPath(path string) (cfg *CGRConfig, err error) { return } err = cfg.checkConfigSanity() - conReqs = NewConReqs(cfg.generalCfg.ConcurrentRequests, cfg.generalCfg.ConcurrentStrategy) return } diff --git a/utils/concreqs_bijson_codec.go b/utils/concreqs_bijson_codec.go new file mode 100644 index 000000000..74301dc81 --- /dev/null +++ b/utils/concreqs_bijson_codec.go @@ -0,0 +1,207 @@ +/* +Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments +Copyright (C) ITsysCOM GmbH + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see +*/ + +// Most of the logic follows standard library implementation in this file +package utils + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "sync" + + "github.com/cenkalti/rpc2" +) + +type concReqsBiJSONCoded struct { + dec *json.Decoder // for reading JSON values + enc *json.Encoder // for writing JSON values + c io.Closer + + // temporary work space + msg message + serverRequest serverRequest + clientResponse clientResponse + + // JSON-RPC clients can use arbitrary json values as request IDs. + // Package rpc expects uint64 request IDs. + // We assign uint64 sequence numbers to incoming requests + // but save the original request ID in the pending map. + // When rpc responds, we use the sequence number in + // the response to find the original request ID. + mutex sync.Mutex // protects seq, pending + pending map[uint64]*json.RawMessage + seq uint64 +} + +// NewConcReqsBiJSONCoded returns a new rpc2.Codec using JSON-RPC on conn. +func NewConcReqsBiJSONCoded(conn io.ReadWriteCloser) rpc2.Codec { + return &concReqsBiJSONCoded{ + dec: json.NewDecoder(conn), + enc: json.NewEncoder(conn), + c: conn, + pending: make(map[uint64]*json.RawMessage), + } +} + +// serverRequest and clientResponse combined +type message struct { + Method string `json:"method"` + Params *json.RawMessage `json:"params"` + Id *json.RawMessage `json:"id"` + Result *json.RawMessage `json:"result"` + Error interface{} `json:"error"` +} + +type clientResponse struct { + Id uint64 `json:"id"` + Result *json.RawMessage `json:"result"` + Error interface{} `json:"error"` +} + +type clientRequest struct { + Method string `json:"method"` + Params []interface{} `json:"params"` + Id *uint64 `json:"id"` +} + +func (c *concReqsBiJSONCoded) ReadHeader(req *rpc2.Request, resp *rpc2.Response) error { + if err := ConReqs.Allocate(); err != nil { + return err + } + c.msg = message{} + if err := c.dec.Decode(&c.msg); err != nil { + return err + } + + if c.msg.Method != "" { + // request comes to server + c.serverRequest.Id = c.msg.Id + c.serverRequest.Method = c.msg.Method + c.serverRequest.Params = c.msg.Params + + req.Method = c.serverRequest.Method + + // JSON request id can be any JSON value; + // RPC package expects uint64. Translate to + // internal uint64 and save JSON on the side. + if c.serverRequest.Id == nil { + // Notification + } else { + c.mutex.Lock() + c.seq++ + c.pending[c.seq] = c.serverRequest.Id + c.serverRequest.Id = nil + req.Seq = c.seq + c.mutex.Unlock() + } + } else { + // response comes to client + err := json.Unmarshal(*c.msg.Id, &c.clientResponse.Id) + if err != nil { + return err + } + c.clientResponse.Result = c.msg.Result + c.clientResponse.Error = c.msg.Error + + resp.Error = "" + resp.Seq = c.clientResponse.Id + if c.clientResponse.Error != nil || c.clientResponse.Result == nil { + x, ok := c.clientResponse.Error.(string) + if !ok { + return fmt.Errorf("invalid error %v", c.clientResponse.Error) + } + if x == "" { + x = "unspecified error" + } + resp.Error = x + } + } + return nil +} + +func (c *concReqsBiJSONCoded) ReadRequestBody(x interface{}) error { + if x == nil { + return nil + } + if c.serverRequest.Params == nil { + return errMissingParams + } + var params *[]interface{} + switch x := x.(type) { + case *[]interface{}: + params = x + default: + params = &[]interface{}{x} + } + return json.Unmarshal(*c.serverRequest.Params, params) +} + +func (c *concReqsBiJSONCoded) ReadResponseBody(x interface{}) error { + if x == nil { + return nil + } + return json.Unmarshal(*c.clientResponse.Result, x) +} + +func (c *concReqsBiJSONCoded) WriteRequest(r *rpc2.Request, param interface{}) error { + req := &clientRequest{Method: r.Method} + switch param := param.(type) { + case []interface{}: + req.Params = param + default: + req.Params = []interface{}{param} + } + if r.Seq == 0 { + // Notification + req.Id = nil + } else { + seq := r.Seq + req.Id = &seq + } + return c.enc.Encode(req) +} + +func (c *concReqsBiJSONCoded) WriteResponse(r *rpc2.Response, x interface{}) error { + defer ConReqs.Deallocate() + c.mutex.Lock() + b, ok := c.pending[r.Seq] + if !ok { + c.mutex.Unlock() + return errors.New("invalid sequence number in response") + } + delete(c.pending, r.Seq) + c.mutex.Unlock() + + if b == nil { + // Invalid request so no id. Use JSON null. + b = &null + } + resp := serverResponse{Id: b} + if r.Error == "" { + resp.Result = x + } else { + resp.Error = r.Error + } + return c.enc.Encode(resp) +} + +func (c *concReqsBiJSONCoded) Close() error { + return c.c.Close() +} diff --git a/config/concureqs.go b/utils/concureqs.go similarity index 84% rename from config/concureqs.go rename to utils/concureqs.go index 34f62c305..ebccb6422 100644 --- a/config/concureqs.go +++ b/utils/concureqs.go @@ -16,14 +16,14 @@ You should have received a copy of the GNU General Public License along with this program. If not, see */ -package config +package utils import ( "fmt" - - "github.com/cgrates/cgrates/utils" ) +var ConReqs *ConcReqs + type ConcReqs struct { aReqs chan struct{} nAReqs int @@ -31,30 +31,34 @@ type ConcReqs struct { } func NewConReqs(reqs int, strategy string) *ConcReqs { - return &ConcReqs{ + cR := &ConcReqs{ aReqs: make(chan struct{}, reqs), nAReqs: reqs, strategy: strategy, } + for i := 0; i < reqs; i++ { + cR.aReqs <- struct{}{} + } + return cR } -func (cR *ConcReqs) VerifyAndGet() (err error) { +func (cR *ConcReqs) Allocate() (err error) { if cR.nAReqs == 0 { return } switch cR.strategy { - case utils.MetaBusy: + case MetaBusy: if len(cR.aReqs) == 0 { return fmt.Errorf("denying request due to maximum active requests reached") } fallthrough - case utils.MetaQueue: + case MetaQueue: <-cR.aReqs // get from channel } return } -func (cR *ConcReqs) Putback() { +func (cR *ConcReqs) Deallocate() { if cR.nAReqs == 0 { return } diff --git a/utils/concureqs_gob_codec.go b/utils/concureqs_gob_codec.go new file mode 100644 index 000000000..9f3757bb8 --- /dev/null +++ b/utils/concureqs_gob_codec.go @@ -0,0 +1,89 @@ +/* +Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments +Copyright (C) ITsysCOM GmbH + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see +*/ + +// Most of the logic follows standard library implementation in this file +package utils + +import ( + "bufio" + "encoding/gob" + "io" + "log" + "net/rpc" +) + +type concReqsGobServerCodec struct { + rwc io.ReadWriteCloser + dec *gob.Decoder + enc *gob.Encoder + encBuf *bufio.Writer + closed bool +} + +func NewConcReqsGobServerCodec(conn io.ReadWriteCloser) rpc.ServerCodec { + buf := bufio.NewWriter(conn) + return &concReqsGobServerCodec{ + rwc: conn, + dec: gob.NewDecoder(conn), + enc: gob.NewEncoder(buf), + encBuf: buf, + } +} + +func (c *concReqsGobServerCodec) ReadRequestHeader(r *rpc.Request) error { + if err := ConReqs.Allocate(); err != nil { + return err + } + return c.dec.Decode(r) +} + +func (c *concReqsGobServerCodec) ReadRequestBody(body interface{}) error { + return c.dec.Decode(body) +} + +func (c *concReqsGobServerCodec) WriteResponse(r *rpc.Response, body interface{}) (err error) { + defer ConReqs.Deallocate() + if err = c.enc.Encode(r); err != nil { + if c.encBuf.Flush() == nil { + // Gob couldn't encode the header. Should not happen, so if it does, + // shut down the connection to signal that the connection is broken. + log.Println("rpc: gob error encoding response:", err) + c.Close() + } + return + } + if err = c.enc.Encode(body); err != nil { + if c.encBuf.Flush() == nil { + // Was a gob problem encoding the body but the header has been written. + // Shut down the connection to signal that the connection is broken. + log.Println("rpc: gob error encoding body:", err) + c.Close() + } + return + } + return c.encBuf.Flush() +} + +func (c *concReqsGobServerCodec) Close() error { + if c.closed { + // Only call c.rwc.Close once; otherwise the semantics are undefined. + return nil + } + c.closed = true + return c.rwc.Close() +} diff --git a/utils/concureqs_json_codec.go b/utils/concureqs_json_codec.go new file mode 100644 index 000000000..5de8bc455 --- /dev/null +++ b/utils/concureqs_json_codec.go @@ -0,0 +1,124 @@ +/* +Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments +Copyright (C) ITsysCOM GmbH + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see +*/ + +// Most of the logic follows standard library implementation in this file +package utils + +import ( + "encoding/json" + "errors" + "io" + "net/rpc" + "sync" +) + +type concReqsServerCodec struct { + dec *json.Decoder // for reading JSON values + enc *json.Encoder // for writing JSON values + c io.Closer + + // temporary work space + req serverRequest + + // JSON-RPC clients can use arbitrary json values as request IDs. + // Package rpc expects uint64 request IDs. + // We assign uint64 sequence numbers to incoming requests + // but save the original request ID in the pending map. + // When rpc responds, we use the sequence number in + // the response to find the original request ID. + mutex sync.Mutex // protects seq, pending + seq uint64 + pending map[uint64]*json.RawMessage +} + +// NewConcReqsServerCodec returns a new rpc.ServerCodec using JSON-RPC on conn. +func NewConcReqsServerCodec(conn io.ReadWriteCloser) rpc.ServerCodec { + return &concReqsServerCodec{ + dec: json.NewDecoder(conn), + enc: json.NewEncoder(conn), + c: conn, + pending: make(map[uint64]*json.RawMessage), + } +} + +func (c *concReqsServerCodec) ReadRequestHeader(r *rpc.Request) error { + if err := ConReqs.Allocate(); err != nil { + return err + } + c.req.reset() + if err := c.dec.Decode(&c.req); err != nil { + return err + } + r.ServiceMethod = c.req.Method + + // JSON request id can be any JSON value; + // RPC package expects uint64. Translate to + // internal uint64 and save JSON on the side. + c.mutex.Lock() + c.seq++ + c.pending[c.seq] = c.req.Id + c.req.Id = nil + r.Seq = c.seq + c.mutex.Unlock() + + return nil +} + +func (c *concReqsServerCodec) ReadRequestBody(x interface{}) error { + if x == nil { + return nil + } + if c.req.Params == nil { + return errMissingParams + } + // JSON params is array value. + // RPC params is struct. + // Unmarshal into array containing struct for now. + // Should think about making RPC more general. + var params [1]interface{} + params[0] = x + return json.Unmarshal(*c.req.Params, ¶ms) +} + +func (c *concReqsServerCodec) WriteResponse(r *rpc.Response, x interface{}) error { + defer ConReqs.Deallocate() + c.mutex.Lock() + b, ok := c.pending[r.Seq] + if !ok { + c.mutex.Unlock() + return errors.New("invalid sequence number in response") + } + delete(c.pending, r.Seq) + c.mutex.Unlock() + + if b == nil { + // Invalid request so no id. Use JSON null. + b = &null + } + resp := serverResponse{Id: b} + if r.Error == "" { + resp.Result = x + } else { + resp.Error = r.Error + } + return c.enc.Encode(resp) +} + +func (c *concReqsServerCodec) Close() error { + return c.c.Close() +} diff --git a/utils/json_codec.go b/utils/json_codec.go index 07c9f11fc..6cdcc9370 100644 --- a/utils/json_codec.go +++ b/utils/json_codec.go @@ -109,6 +109,9 @@ func (c *jsonServerCodec) ReadRequestHeader(r *rpc.Request) error { } func (c *jsonServerCodec) ReadRequestBody(x interface{}) error { + if err := ConReqs.Allocate(); err != nil { + return err + } if x == nil { return nil } @@ -137,6 +140,7 @@ func (c *jsonServerCodec) ReadRequestBody(x interface{}) error { var null = json.RawMessage([]byte("null")) func (c *jsonServerCodec) WriteResponse(r *rpc.Response, x interface{}) error { + defer ConReqs.Deallocate() c.mutex.Lock() b, ok := c.pending[r.Seq] if !ok { diff --git a/utils/server.go b/utils/server.go index 764471db8..ea11e7b49 100644 --- a/utils/server.go +++ b/utils/server.go @@ -31,7 +31,6 @@ import ( "net/http" "net/http/pprof" "net/rpc" - "net/rpc/jsonrpc" "net/url" "reflect" "strings" @@ -39,7 +38,6 @@ import ( "time" "github.com/cenkalti/rpc2" - rpc2_jsonrpc "github.com/cenkalti/rpc2/jsonrpc" "golang.org/x/net/websocket" ) @@ -174,13 +172,11 @@ func (s *Server) ServeJSON(addr string, exitChan chan bool) { } continue } - //utils.Logger.Info(fmt.Sprintf(" New incoming connection: %v", conn.RemoteAddr())) if s.isDispatched { go rpc.ServeCodec(NewCustomJSONServerCodec(conn)) } else { - go jsonrpc.ServeConn(conn) + go rpc.ServeCodec(NewConcReqsServerCodec(conn)) } - } } @@ -216,9 +212,7 @@ func (s *Server) ServeGOB(addr string, exitChan chan bool) { } continue } - - //utils.Logger.Info(fmt.Sprintf(" New incoming connection: %v", conn.RemoteAddr())) - go rpc.ServeConn(conn) + go rpc.ServeCodec(NewConcReqsGobServerCodec(conn)) } } @@ -281,7 +275,7 @@ func (s *Server) ServeHTTP(addr string, jsonRPCURL string, wsRPCURL string, if s.isDispatched { rpc.ServeCodec(NewCustomJSONServerCodec(ws)) } else { - jsonrpc.ServeConn(ws) + rpc.ServeCodec(NewConcReqsServerCodec(ws)) } }) if useBasicAuth { @@ -333,7 +327,7 @@ func (s *Server) ServeBiJSON(addr string, onConn func(*rpc2.Client), onDis func( log.Fatal(err) return // stop if we get Accept error } - go s.birpcSrv.ServeCodec(rpc2_jsonrpc.NewJSONCodec(conn)) + go s.birpcSrv.ServeCodec(NewConcReqsBiJSONCoded(conn)) } }(lBiJSON) <-s.stopbiRPCServer // wait until server is stoped to close the listener @@ -378,7 +372,7 @@ func (r *rpcRequest) Close() error { // Call invokes the RPC request, waits for it to complete, and returns the results. func (r *rpcRequest) Call() io.Reader { - go jsonrpc.ServeConn(r) + go rpc.ServeCodec(NewConcReqsServerCodec(r)) <-r.done return r.rw } @@ -460,8 +454,7 @@ func (s *Server) ServeGOBTLS(addr, serverCrt, serverKey, caCert string, } continue } - //utils.Logger.Info(fmt.Sprintf(" New incoming connection: %v", conn.RemoteAddr())) - go rpc.ServeConn(conn) + go rpc.ServeCodec(NewConcReqsGobServerCodec(conn)) } } @@ -505,7 +498,7 @@ func (s *Server) ServeJSONTLS(addr, serverCrt, serverKey, caCert string, if s.isDispatched { go rpc.ServeCodec(NewCustomJSONServerCodec(conn)) } else { - go jsonrpc.ServeConn(conn) + go rpc.ServeCodec(NewConcReqsServerCodec(conn)) } } } @@ -540,7 +533,7 @@ func (s *Server) ServeHTTPTLS(addr, serverCrt, serverKey, caCert string, serverP if s.isDispatched { rpc.ServeCodec(NewCustomJSONServerCodec(ws)) } else { - jsonrpc.ServeConn(ws) + rpc.ServeCodec(NewConcReqsServerCodec(ws)) } }) if useBasicAuth {