Add support for limited concurrent requests

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")
 	}

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("<CoreS> starting version <%s><%s>", vers, goVers))
 	cfg.LazySanityCheck()
 

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
 }
 

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 <http://www.gnu.org/licenses/>
+*/
+
+// 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()
+}

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 <http://www.gnu.org/licenses/>
 */
 
-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
 	}

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 <http://www.gnu.org/licenses/>
+*/
+
+// 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()
+}

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 <http://www.gnu.org/licenses/>
+*/
+
+// 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, &params)
+}
+
+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()
+}

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 {

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("<CGRServer> 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("<CGRServer> 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("<CGRServer> 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 {