func (s *Server) setContext(t *task, id string) bool {
	base, params, err := s.dectx(s.newctx(), t.hreq.method, t.hreq.params)
	t.hreq.params = params
	if err != nil {
		t.err = Errorf(code.InternalError, "invalid request context: %v", err)
		return false
	}
	if err := s.ckreq(base, t.hreq); err != nil {
		t.err = err
		return false
	}
	t.ctx = context.WithValue(base, inboundRequestKey{}, t.hreq)
	if id != "" {
		ctx, cancel := context.WithCancel(t.ctx)
		s.used[id] = cancel
		t.ctx = ctx
	}
	return true
}
setContext constructs and attaches a request context to t, and reports
whether this succeeded.
package jrpc2
import (
	"container/list"
	"context"
	"encoding/json"
	"errors"
	"io"
	"strconv"
	"strings"
	"sync"
	"time"
	"github.com/creachadair/jrpc2/channel"
	"github.com/creachadair/jrpc2/code"
	"github.com/creachadair/jrpc2/metrics"
	"golang.org/x/sync/semaphore"
)
type Server struct {
	wg  sync.WaitGroup       
	mux Assigner             
	sem *semaphore.Weighted  
	allow1  bool                          
	allowP  bool                          
	log     func(string, ...interface{})  
	rpcLog  RPCLogger                     
	newctx  func() context.Context        
	dectx   decoder                       
	ckreq   verifier                      
	expctx  bool                          
	metrics *metrics.M                    
	start   time.Time                     
	builtin bool                          
	mu *sync.Mutex  
	nbar sync.WaitGroup   
	err  error            
	work *sync.Cond       
	inq  *list.List       
	ch   channel.Channel  
	used map[string]context.CancelFunc
	call   map[string]*Response
	callID int64
}
func NewServer(mux Assigner, opts *ServerOptions) *Server {
	if mux == nil {
		panic("nil assigner")
	}
	dc, exp := opts.decodeContext()
	s := &Server{
		mux:     mux,
		sem:     semaphore.NewWeighted(opts.concurrency()),
		allow1:  opts.allowV1(),
		allowP:  opts.allowPush(),
		log:     opts.logFunc(),
		rpcLog:  opts.rpcLog(),
		newctx:  opts.newContext(),
		dectx:   dc,
		ckreq:   opts.checkRequest(),
		expctx:  exp,
		mu:      new(sync.Mutex),
		metrics: opts.metrics(),
		start:   opts.startTime(),
		builtin: opts.allowBuiltin(),
		inq:     list.New(),
		used:    make(map[string]context.CancelFunc),
		call:    make(map[string]*Response),
		callID:  1,
	}
	s.work = sync.NewCond(s.mu)
	return s
}
func (s *Server) Start(c channel.Channel) *Server {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.ch != nil {
		panic("server is already running")
	}
	s.ch = c
	if s.start.IsZero() {
		s.start = time.Now().In(time.UTC)
	}
	s.err = nil
	s.wg.Add(2)
	go func() { defer s.wg.Done(); s.read(c) }()
	go func() { defer s.wg.Done(); s.serve() }()
	return s
}
func (s *Server) serve() {
	for {
		next, err := s.nextRequest()
		if err != nil {
			s.log("Error reading from client: %v", err)
			return
		}
		s.wg.Add(1)
		go func() {
			defer s.wg.Done()
			next()
		}()
	}
}
func (s *Server) nextRequest() (func() error, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	for s.ch != nil && s.inq.Len() == 0 {
		s.work.Wait()
	}
	if s.ch == nil && s.inq.Len() == 0 {
		return nil, s.err
	}
	ch := s.ch  
	next := s.inq.Remove(s.inq.Front()).(jmessages)
	s.log("Dequeued request batch of length %d (qlen=%d)", len(next), s.inq.Len())
	return s.dispatch(next, ch), nil
}
func (s *Server) waitForBarrier(n int) {
	s.mu.Unlock()
	defer s.mu.Lock()
	s.nbar.Wait()
	s.nbar.Add(n)
}
func (s *Server) dispatch(next jmessages, ch sender) func() error {
	start := time.Now()
	tasks := s.checkAndAssign(next)
	last := len(tasks) - 1
	s.waitForBarrier(tasks.numValidNotifications())
	return func() error {
		var wg sync.WaitGroup
		for i, t := range tasks {
			if t.err != nil {
				continue  
			}
			t := t
			wg.Add(1)
			run := func() {
				defer wg.Done()
				if t.hreq.IsNotification() {
					defer s.nbar.Done()
				}
				t.val, t.err = s.invoke(t.ctx, t.m, t.hreq)
			}
			if i < last {
				go run()
			} else {
				run()
			}
		}
		wg.Wait()
		return s.deliver(tasks.responses(s.rpcLog), ch, time.Since(start))
	}
}
func (s *Server) deliver(rsps jmessages, ch sender, elapsed time.Duration) error {
	if len(rsps) == 0 {
		return nil
	}
	s.log("Completed %d requests [%v elapsed]", len(rsps), elapsed)
	s.mu.Lock()
	defer s.mu.Unlock()
	for _, rsp := range rsps {
		if rsp.err == nil {
			s.cancel(string(rsp.ID))
		}
	}
	nw, err := encode(ch, rsps)
	s.metrics.CountAndSetMax("rpc.bytesWritten", int64(nw))
	return err
}
func (s *Server) checkAndAssign(next jmessages) tasks {
	var ts tasks
	for _, req := range next {
		fid := fixID(req.ID)
		t := &task{
			hreq:  &Request{id: fid, method: req.M, params: req.P},
			batch: req.batch,
		}
		id := string(fid)
		if req.err != nil {
			t.err = req.err  
		} else if !req.isRequestOrNotification() && s.call[id] != nil {
			rsp := s.call[id]
			delete(s.call, id)
			rsp.ch <- req
			continue  
		} else if id != "" && s.used[id] != nil {
			t.err = Errorf(code.InvalidRequest, "duplicate request id %q", id)
		} else if !s.versionOK(req.V) {
			t.err = ErrInvalidVersion
		} else if req.M == "" {
			t.err = errEmptyMethod
		} else if s.setContext(t, id) {
			t.m = s.assign(t.ctx, req.M)
			if t.m == nil {
				t.err = Errorf(code.MethodNotFound, "no such method %q", req.M)
			}
		}
		if t.err != nil {
			s.log("Request check error for %q (params %q): %v", req.M, string(req.P), t.err)
			s.metrics.Count("rpc.errors", 1)
		}
		ts = append(ts, t)
	}
	return ts
}