func (s *RpcServer) serveRequest(ctx context.Context, codec ServerCodec, singleShot bool, options CodecOption) error {
	var pend sync.WaitGroup
	defer func() {
		if err := recover(); err != nil {
			const size = 64 << 10
			buf := make([]byte, size)
			buf = buf[:runtime.Stack(buf, false)]
			log.Error(string(buf))
		}
		s.codecsMu.Lock()
		s.codecs.Remove(codec)
		s.codecsMu.Unlock()
	}()
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	if options&OptionSubscriptions == OptionSubscriptions {
		ctx = context.WithValue(ctx, notifierKey{}, newNotifier(codec))
	}
	s.codecsMu.Lock()
	if atomic.LoadInt32(&s.run) != 1 {  
		s.codecsMu.Unlock()
		return &shutdownError{}
	}
	s.codecs.Add(codec)
	s.codecsMu.Unlock()
	for atomic.LoadInt32(&s.run) == 1 {
		reqs, batch, err := s.readRequest(codec)
		if err != nil {
			if err.Error() != "EOF" {
				log.Debug(fmt.Sprintf("read error %v\n", err))
				_ = codec.Write(codec.CreateErrorResponse(nil, err))
			}
			pend.Wait()
			return nil
		}
		if atomic.LoadInt32(&s.run) != 1 {
			err = &shutdownError{}
			if batch {
				resps := make([]interface{}, len(reqs))
				for i, r := range reqs {
					resps[i] = codec.CreateErrorResponse(&r.id, err)
				}
				codec.Write(resps)
			} else {
				codec.Write(codec.CreateErrorResponse(&reqs[0].id, err))
			}
			return nil
		}
		if singleShot {
			if batch {
				s.execBatch(ctx, codec, reqs)
			} else {
				s.exec(ctx, codec, reqs[0])
			}
			return nil
		}
		pend.Add(1)
		go func(reqs []*serverRequest, batch bool) {
			defer pend.Done()
			if batch {
				s.execBatch(ctx, codec, reqs)
			} else {
				s.exec(ctx, codec, reqs[0])
			}
		}(reqs, batch)
	}
	return nil
}
serveRequest will reads requests from the codec, calls the RPC callback and
writes the response to the given codec.

If singleShot is true it will process a single request, otherwise it will handle
requests until the codec returns an error when reading a request (in most cases
an EOF). It executes requests in parallel when singleShot is false.
package server
import (
	"crypto/sha256"
	"crypto/subtle"
	"encoding/base64"
	"fmt"
	"io"
	"net"
	"net/http"
	"reflect"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/Qitmeer/qitmeer/log"
	mapset "github.com/deckarep/golang-set"
	"golang.org/x/net/context"
	"github.com/Qitmeer/qitmeer/common/util"
)
type API struct {
	NameSpace string       
	Service   interface{}  
}
type RpcServer struct {
	run        int32
	wg         util.WaitGroupWrapper
	quit       chan int
	statusLock sync.RWMutex
	config *Config
	rpcSvcRegistry serviceRegistry
	codecsMu sync.Mutex
	codecs   mapset.Set
	authsha                [sha256.Size]byte
	numClients             int32
	statusLines            map[int]string
	requestProcessShutdown chan struct{}
}
type service struct {
	svcNamespace  string         
	svcType       reflect.Type   
	callbacks     callbacks      
	subscriptions subscriptions  
}
type callback struct {
	receiver    reflect.Value   
	method      reflect.Method  
	argTypes    []reflect.Type  
	hasCtx      bool            
	errPos      int             
	isSubscribe bool            
}
type serviceRegistry map[string]*service
type callbacks map[string]*callback
type subscriptions map[string]*callback
type rpcRequest struct {
	service  string
	method   string
	id       interface{}
	isPubSub bool
	params   interface{}
	err      Error  
}
type serverRequest struct {
	id            interface{}
	svcname       string
	callb         *callback
	args          []reflect.Value
	isUnsubscribe bool
	err           Error
}
func NewRPCServer(cfg *Config) (*RpcServer, error) {
	rpc := RpcServer{
		config: cfg,
		rpcSvcRegistry: make(serviceRegistry),
		codecs:         mapset.NewSet(),
		statusLines:            make(map[int]string),
		requestProcessShutdown: make(chan struct{}),
		quit:                   make(chan int),
	}
	if cfg.RPCUser != "" && cfg.RPCPass != "" {
		login := cfg.RPCUser + ":" + cfg.RPCPass
		auth := "Basic " +
			base64.StdEncoding.EncodeToString([]byte(login))
		rpc.authsha = sha256.Sum256([]byte(auth))
	}
	return &rpc, nil
}
func (s *RpcServer) Start() error {
	s.run = 1
	return nil
}
func (s *RpcServer) Stop() {
	if atomic.CompareAndSwapInt32(&s.run, 1, 0) {
		log.Debug("RPC Server is stopping")
		s.codecsMu.Lock()
		defer s.codecsMu.Unlock()
		s.codecs.Each(func(c interface{}) bool {
			c.(ServerCodec).Close()
			return true
		})
	}
}
const (
	rpcAuthTimeoutSeconds = 10
)
func (s *RpcServer) HandleFunc(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Connection", "close")
	w.Header().Set("Content-Type", "application/json")
	r.Close = true
	if s.limitConnections(w, r.RemoteAddr) {
		return
	}
	s.incrementClients()
	defer s.decrementClients()
	_, err := s.checkAuth(r, true)
	if err != nil {
		jsonAuthFail(w)
		return
	}
	s.jsonRPCRead(w, r)
}
func (s *RpcServer) startHTTP(listenAddrs []string) error {
	rpcServeMux := http.NewServeMux()
	httpServer := &http.Server{
		Handler: rpcServeMux,
		ReadTimeout: time.Second * rpcAuthTimeoutSeconds,
	}
	rpcServeMux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Connection", "close")
		w.Header().Set("Content-Type", "application/json")
		r.Close = true
		if s.limitConnections(w, r.RemoteAddr) {
			return
		}
		s.incrementClients()
		defer s.decrementClients()
		_, err := s.checkAuth(r, true)
		if err != nil {
			jsonAuthFail(w)
			return
		}
		s.jsonRPCRead(w, r)
	})
	listeners, err := parseListeners(s.config, listenAddrs)
	if err != nil {
		return err
	}
	for _, listener := range listeners {
		s.wg.Add(1)
		go func(listener net.Listener) {
			log.Info("RPC server listening on ", "addr", listener.Addr())
			_ = httpServer.Serve(listener)
			log.Trace("RPC listener done for %s", listener.Addr())
			s.wg.Done()
		}(listener)
	}
	return nil
}
func (s *RpcServer) limitConnections(w http.ResponseWriter, remoteAddr string) bool {
	if int(atomic.LoadInt32(&s.numClients)+1) > s.config.RPCMaxClients {
		log.Info("RPC clients exceeded", "max", s.config.RPCMaxClients,
			"client", remoteAddr)
		http.Error(w, "503 Too busy.  Try again later.",
			http.StatusServiceUnavailable)
		return true
	}
	return false
}
func (s *RpcServer) incrementClients() {
	atomic.AddInt32(&s.numClients, 1)
}
func (s *RpcServer) decrementClients() {
	atomic.AddInt32(&s.numClients, -1)
}
func (s *RpcServer) checkAuth(r *http.Request, require bool) (bool, error) {
	authhdr := r.Header["Authorization"]
	if len(authhdr) <= 0 {
		if require {
			log.Warn("RPC authentication failure", " from ", r.RemoteAddr,
				" error ", "no authorization header")
			return false, fmt.Errorf("auth failure")
		}
		return false, nil
	}
	authsha := sha256.Sum256([]byte(authhdr[0]))
	cmp := subtle.ConstantTimeCompare(authsha[:], s.authsha[:])
	if cmp == 1 {
		return true, nil
	}
	log.Warn("RPC authentication failure", "from", r.RemoteAddr)
	return false, fmt.Errorf("auth failure")
}
func jsonAuthFail(w http.ResponseWriter) {
	w.Header().Add("WWW-Authenticate", `Basic realm="nox RPC"`)
	http.Error(w, "401 Unauthorized.", http.StatusUnauthorized)
}
type CodecOption int
const (
	OptionMethodInvocation CodecOption = 1 << iota
	OptionSubscriptions = 1 << iota  
)
func (s *RpcServer) jsonRPCRead(w http.ResponseWriter, r *http.Request) {
	if atomic.LoadInt32(&s.run) != 1 {  
		log.Info("RpcServer jsonRPCRead s.run ")
	}
	if emptyRequest(r) {
		log.Trace("Discard empty request", "from", r.RemoteAddr, "request", r)
		return
	}
	if code, err := validateRequest(r); err != nil {
		http.Error(w, err.Error(), code)
		return
	}
	ctx := r.Context()
	ctx = context.WithValue(ctx, "remote", r.RemoteAddr)
	ctx = context.WithValue(ctx, "scheme", r.Proto)
	ctx = context.WithValue(ctx, "local", r.Host)
	body := io.LimitReader(r.Body, maxRequestContentLength)
	codec := NewJSONCodec(&httpReadWriteNopCloser{body, w})
	defer codec.Close()
	s.ServeSingleRequest(ctx, codec, OptionMethodInvocation)
}
func (s *RpcServer) ServeSingleRequest(ctx context.Context, codec ServerCodec, options CodecOption) {
	_ = s.serveRequest(ctx, codec, true, options)
}