func (d *DefaultServerDispatcher) messagePump() {
	var clientID string
	var ok bool
	var rdy bool
	var clientQueue RequestQueue
	clientReadyMap := map[string]bool{}  
	for {
		select {
		case clientID, ok = <-d.requestChannel:
			if !ok {
				d.queueMap.Init()
				d.requestChannel = nil
				log.Info("stopped processing requests")
				return
			}
			clientQueue, ok = d.queueMap.Get(clientID)
			if !ok {
				delete(clientReadyMap, clientID)
				rdy = false
				break
			}
			rdy, ok = clientReadyMap[clientID]
			if !ok {
				rdy = true
				clientReadyMap[clientID] = rdy
			}
		case clientID = <-d.readyForDispatch:
			clientQueue, rdy = d.queueMap.Get(clientID)
			if rdy {
				clientReadyMap[clientID] = rdy
			}
		}
		if rdy && !clientQueue.IsEmpty() {
			d.dispatchNextRequest(clientID)
			rdy = false
			clientReadyMap[clientID] = rdy
		}
	}
}
requestPump processes new outgoing requests for each client and makes sure they are processed sequentially.
This method is executed by a dedicated coroutine as soon as the server is started and runs indefinitely.
package ocppj
import (
	"fmt"
	"sync"
	"time"
	"github.com/lorenzodonini/ocpp-go/ocpp"
	"github.com/lorenzodonini/ocpp-go/ws"
)
type ClientDispatcher interface {
	Start()
	SetTimeout(timeout time.Duration)
	IsRunning() bool
	IsPaused() bool
	SendRequest(req RequestBundle) error
	CompleteRequest(requestID string)
	SetOnRequestCanceled(cb CanceledRequestHandler)
	SetNetworkClient(client ws.WsClient)
	SetPendingRequestState(stateHandler ClientState)
	Stop()
	Pause()
	Resume()
}
type pendingRequest struct {
	request   ocpp.Request
	startTime time.Time
}
type CanceledRequestHandler func(id string, action string, request ocpp.Request)
type DefaultClientDispatcher struct {
	requestQueue        RequestQueue
	requestChannel      chan bool
	readyForDispatch    chan bool
	pendingRequestState ClientState
	network             ws.WsClient
	mutex               sync.RWMutex
	onRequestCancel     CanceledRequestHandler
	timer               *time.Timer
	paused              bool
	timeout             time.Duration
}
const defaultTimeoutTick = 24 * time.Hour
const defaultMessageTimeout = 30 * time.Second
func NewDefaultClientDispatcher(queue RequestQueue) *DefaultClientDispatcher {
	return &DefaultClientDispatcher{
		requestQueue:        queue,
		requestChannel:      nil,
		readyForDispatch:    make(chan bool, 1),
		pendingRequestState: NewClientState(),
		timeout:             defaultMessageTimeout,
	}
}
func (d *DefaultClientDispatcher) SetOnRequestCanceled(cb CanceledRequestHandler) {
	d.onRequestCancel = cb
}
func (d *DefaultClientDispatcher) SetTimeout(timeout time.Duration) {
	d.timeout = timeout
}
func (d *DefaultClientDispatcher) Start() {
	d.requestChannel = make(chan bool, 1)
	d.timer = time.NewTimer(defaultTimeoutTick)  
	go d.messagePump()
}
func (d *DefaultClientDispatcher) IsRunning() bool {
	d.mutex.Lock()
	defer d.mutex.Unlock()
	return d.requestChannel != nil
}
func (d *DefaultClientDispatcher) IsPaused() bool {
	d.mutex.Lock()
	defer d.mutex.Unlock()
	return d.paused
}
func (d *DefaultClientDispatcher) Stop() {
	d.mutex.Lock()
	defer d.mutex.Unlock()
	close(d.requestChannel)
}
func (d *DefaultClientDispatcher) SetNetworkClient(client ws.WsClient) {
	d.network = client
}
func (d *DefaultClientDispatcher) SetPendingRequestState(state ClientState) {
	d.pendingRequestState = state
}
func (d *DefaultClientDispatcher) SendRequest(req RequestBundle) error {
	if d.network == nil {
		return fmt.Errorf("cannot SendRequest, no network client was set")
	}
	if err := d.requestQueue.Push(req); err != nil {
		return err
	}
	d.requestChannel <- true
	return nil
}
func (d *DefaultClientDispatcher) messagePump() {
	rdy := true  
	for {
		select {
		case _, ok := <-d.requestChannel:
			if !ok {
				d.requestQueue.Init()
				d.requestChannel = nil
				return
			}
		case _, ok := <-d.timer.C:
			if !ok {
				continue
			}
			if d.pendingRequestState.HasPendingRequest() {
				el := d.requestQueue.Peek()
				bundle, _ := el.(RequestBundle)
				d.CompleteRequest(bundle.Call.UniqueId)
				if d.onRequestCancel != nil {
					d.onRequestCancel(bundle.Call.UniqueId, bundle.Call.Action, bundle.Call.Payload)
				}
			}
			d.timer.Reset(defaultTimeoutTick)
		case rdy = <-d.readyForDispatch:
		}
		d.mutex.Lock()
		paused := d.paused
		d.mutex.Unlock()
		if paused {
			continue
		}
		if rdy && !d.requestQueue.IsEmpty() {
			d.dispatchNextRequest()
			rdy = false
			if !d.timer.Stop() {
				<-d.timer.C
			}
			d.timer.Reset(d.timeout)
		}
	}
}
func (d *DefaultClientDispatcher) dispatchNextRequest() {
	el := d.requestQueue.Peek()
	bundle, _ := el.(RequestBundle)
	jsonMessage := bundle.Data
	d.pendingRequestState.AddPendingRequest(bundle.Call.UniqueId, bundle.Call.Payload)
	err := d.network.Write(jsonMessage)
	if err != nil {
		d.CompleteRequest(bundle.Call.GetUniqueId())
		if d.onRequestCancel != nil {
			d.onRequestCancel(bundle.Call.UniqueId, bundle.Call.Action, bundle.Call.Payload)
		}
	}
}
func (d *DefaultClientDispatcher) Pause() {
	d.mutex.Lock()
	defer d.mutex.Unlock()
	if !d.timer.Stop() {
		<-d.timer.C
	}
	d.timer.Reset(defaultTimeoutTick)
	d.paused = true
}
func (d *DefaultClientDispatcher) Resume() {
	d.mutex.Lock()
	d.paused = false
	d.mutex.Unlock()
	if d.pendingRequestState.HasPendingRequest() {
		d.timer.Reset(d.timeout)
	} else {
		d.readyForDispatch <- true
	}
}
func (d *DefaultClientDispatcher) CompleteRequest(requestId string) {
	el := d.requestQueue.Peek()
	if el == nil {
		log.Errorf("attempting to pop front of queue, but queue is empty")
		return
	}
	bundle, _ := el.(RequestBundle)
	if bundle.Call.UniqueId != requestId {
		log.Errorf("internal state mismatch: received response for %v but expected response for %v", requestId, bundle.Call.UniqueId)
		return
	}
	d.requestQueue.Pop()
	d.pendingRequestState.DeletePendingRequest(requestId)
	log.Debugf("removed request %v from front of queue", bundle.Call.UniqueId)
	d.readyForDispatch <- true
}
type ServerDispatcher interface {
	Start()
	IsRunning() bool
	SendRequest(clientID string, req RequestBundle) error
	CompleteRequest(clientID string, requestID string)
	SetOnRequestCanceled(cb func(string, string, string, ocpp.Request))
	SetNetworkServer(server ws.WsServer)
	SetPendingRequestState(stateHandler ServerState)
	Stop()
	CreateClient(clientID string)
	DeleteClient(clientID string)
}
type DefaultServerDispatcher struct {
	queueMap            ServerQueueMap
	requestChannel      chan string
	readyForDispatch    chan string
	pendingRequestState ServerState
	onRequestCancel     func(string, string, string, ocpp.Request)
	network             ws.WsServer
	mutex               sync.RWMutex
}
func NewDefaultServerDispatcher(queueMap ServerQueueMap) *DefaultServerDispatcher {
	d := &DefaultServerDispatcher{
		queueMap:         queueMap,
		requestChannel:   nil,
		readyForDispatch: make(chan string, 1),
	}
	d.pendingRequestState = NewServerState(&d.mutex)
	return d
}
func (d *DefaultServerDispatcher) Start() {
	d.requestChannel = make(chan string, 1)
	go d.messagePump()
}
func (d *DefaultServerDispatcher) IsRunning() bool {
	d.mutex.Lock()
	defer d.mutex.Unlock()
	return d.requestChannel != nil
}
func (d *DefaultServerDispatcher) Stop() {
	d.mutex.Lock()
	defer d.mutex.Unlock()
	close(d.requestChannel)
}
func (d *DefaultServerDispatcher) CreateClient(clientID string) {
	_ = d.queueMap.GetOrCreate(clientID)
}
func (d *DefaultServerDispatcher) DeleteClient(clientID string) {
	d.queueMap.Remove(clientID)
	d.requestChannel <- clientID
}
func (d *DefaultServerDispatcher) SetNetworkServer(server ws.WsServer) {
	d.network = server
}
func (d *DefaultServerDispatcher) SetOnRequestCanceled(cb func(string, string, string, ocpp.Request)) {
	d.onRequestCancel = cb
}
func (d *DefaultServerDispatcher) SetPendingRequestState(state ServerState) {
	d.pendingRequestState = state
}
func (d *DefaultServerDispatcher) SendRequest(clientID string, req RequestBundle) error {
	if d.network == nil {
		return fmt.Errorf("cannot send request %v, no network server was set", req.Call.UniqueId)
	}
	q, ok := d.queueMap.Get(clientID)
	if !ok {
		return fmt.Errorf("cannot send request %s, no client %s exists", req.Call.UniqueId, clientID)
	}
	if err := q.Push(req); err != nil {
		return err
	}
	d.requestChannel <- clientID
	return nil
}