Ocpp: cache and re-use initial status (2nd attempt) (#16885)
This commit is contained in:
parent
5827307a65
commit
6f7489af2d
4 changed files with 90 additions and 60 deletions
|
|
@ -158,8 +158,20 @@ func (cp *CP) Setup(meterValues string, meterInterval time.Duration) error {
|
|||
|
||||
// trigger status for all connectors
|
||||
if cp.HasRemoteTriggerFeature {
|
||||
if err := cp.TriggerMessageRequest(0, core.StatusNotificationFeatureName); err != nil {
|
||||
cp.log.WARN.Printf("failed triggering StatusNotification: %v", err)
|
||||
var ok bool
|
||||
|
||||
// apply cached status if available
|
||||
instance.WithChargepointStatusByID(cp.id, func(status *core.StatusNotificationRequest) {
|
||||
if _, err := cp.OnStatusNotification(status); err == nil {
|
||||
ok = true
|
||||
}
|
||||
})
|
||||
|
||||
// only trigger if we don't already have a status
|
||||
if !ok {
|
||||
if err := cp.TriggerMessageRequest(0, core.StatusNotificationFeatureName); err != nil {
|
||||
cp.log.WARN.Printf("failed triggering StatusNotification: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -8,43 +8,22 @@ import (
|
|||
|
||||
"github.com/evcc-io/evcc/util"
|
||||
ocpp16 "github.com/lorenzodonini/ocpp-go/ocpp1.6"
|
||||
"github.com/lorenzodonini/ocpp-go/ocpp1.6/core"
|
||||
)
|
||||
|
||||
type CS struct {
|
||||
mu sync.Mutex
|
||||
log *util.Logger
|
||||
ocpp16.CentralSystem
|
||||
cps map[string]*CP
|
||||
init map[string]*sync.Mutex
|
||||
txnId atomic.Int64
|
||||
type registration struct {
|
||||
mu sync.RWMutex
|
||||
setup sync.RWMutex // serialises chargepoint setup
|
||||
cp *CP // guarded by setup and CS mutexes
|
||||
status *core.StatusNotificationRequest // guarded by mu mutex
|
||||
}
|
||||
|
||||
// Register registers a charge point with the central system.
|
||||
// The charge point identified by id may already be connected in which case initial connection is triggered.
|
||||
func (cs *CS) register(id string, new *CP) error {
|
||||
cs.mu.Lock()
|
||||
defer cs.mu.Unlock()
|
||||
|
||||
cp, ok := cs.cps[id]
|
||||
|
||||
// case 1: charge point neither registered nor physically connected
|
||||
if !ok {
|
||||
cs.cps[id] = new
|
||||
return nil
|
||||
}
|
||||
|
||||
// case 2: duplicate registration of id empty
|
||||
if id == "" {
|
||||
return errors.New("cannot have >1 charge point with empty station id")
|
||||
}
|
||||
|
||||
// case 3: charge point not registered but physically already connected
|
||||
if cp == nil {
|
||||
cs.cps[id] = new
|
||||
new.connect(true)
|
||||
}
|
||||
|
||||
return nil
|
||||
type CS struct {
|
||||
ocpp16.CentralSystem
|
||||
mu sync.Mutex
|
||||
log *util.Logger
|
||||
regs map[string]*registration // guarded by mu mutex
|
||||
txnId atomic.Int64
|
||||
}
|
||||
|
||||
// errorHandler logs error channel
|
||||
|
|
@ -58,38 +37,67 @@ func (cs *CS) ChargepointByID(id string) (*CP, error) {
|
|||
cs.mu.Lock()
|
||||
defer cs.mu.Unlock()
|
||||
|
||||
cp, ok := cs.cps[id]
|
||||
reg, ok := cs.regs[id]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("unknown charge point: %s", id)
|
||||
}
|
||||
if cp == nil {
|
||||
if reg.cp == nil {
|
||||
return nil, fmt.Errorf("charge point not configured: %s", id)
|
||||
}
|
||||
return cp, nil
|
||||
return reg.cp, nil
|
||||
}
|
||||
|
||||
func (cs *CS) WithChargepointStatusByID(id string, fun func(status *core.StatusNotificationRequest)) {
|
||||
cs.mu.Lock()
|
||||
defer cs.mu.Unlock()
|
||||
|
||||
if reg, ok := cs.regs[id]; ok {
|
||||
reg.mu.RLock()
|
||||
if reg.status != nil {
|
||||
fun(reg.status)
|
||||
}
|
||||
reg.mu.RUnlock()
|
||||
}
|
||||
}
|
||||
|
||||
// RegisterChargepoint registers a charge point with the central system of returns an already registered charge point
|
||||
func (cs *CS) RegisterChargepoint(id string, newfun func() *CP, init func(*CP) error) (*CP, error) {
|
||||
cs.mu.Lock()
|
||||
cpmu, ok := cs.init[id]
|
||||
if !ok {
|
||||
cpmu = new(sync.Mutex)
|
||||
cs.init[id] = cpmu
|
||||
|
||||
// prepare shadow state
|
||||
reg, registered := cs.regs[id]
|
||||
if !registered {
|
||||
reg = new(registration)
|
||||
cs.regs[id] = reg
|
||||
}
|
||||
cs.mu.Unlock()
|
||||
|
||||
// serialise on chargepoint id
|
||||
cpmu.Lock()
|
||||
defer cpmu.Unlock()
|
||||
reg.setup.Lock()
|
||||
defer reg.setup.Unlock()
|
||||
|
||||
cp := reg.cp
|
||||
|
||||
cs.mu.Unlock()
|
||||
|
||||
// setup already completed?
|
||||
if cp != nil {
|
||||
// duplicate registration of id empty
|
||||
if id == "" {
|
||||
return nil, errors.New("cannot have >1 charge point with empty station id")
|
||||
}
|
||||
|
||||
// already registered?
|
||||
if cp, err := cs.ChargepointByID(id); err == nil {
|
||||
return cp, nil
|
||||
}
|
||||
|
||||
// first time- registration should not error
|
||||
cp := newfun()
|
||||
if err := cs.register(id, cp); err != nil {
|
||||
return nil, err
|
||||
// first time- create the charge point
|
||||
cp = newfun()
|
||||
|
||||
cs.mu.Lock()
|
||||
reg.cp = cp
|
||||
cs.mu.Unlock()
|
||||
|
||||
if registered {
|
||||
cp.connect(true)
|
||||
}
|
||||
|
||||
return cp, init(cp)
|
||||
|
|
@ -101,12 +109,12 @@ func (cs *CS) NewChargePoint(chargePoint ocpp16.ChargePointConnection) {
|
|||
defer cs.mu.Unlock()
|
||||
|
||||
// check for configured charge point
|
||||
cp, ok := cs.cps[chargePoint.ID()]
|
||||
reg, ok := cs.regs[chargePoint.ID()]
|
||||
if ok {
|
||||
cs.log.DEBUG.Printf("charge point connected: %s", chargePoint.ID())
|
||||
|
||||
// trigger initial connection if charge point is already setup
|
||||
if cp != nil {
|
||||
if cp := reg.cp; cp != nil {
|
||||
cp.connect(true)
|
||||
}
|
||||
|
||||
|
|
@ -114,15 +122,16 @@ func (cs *CS) NewChargePoint(chargePoint ocpp16.ChargePointConnection) {
|
|||
}
|
||||
|
||||
// check for configured anonymous charge point
|
||||
cp, ok = cs.cps[""]
|
||||
if ok && cp != nil {
|
||||
reg, ok = cs.regs[""]
|
||||
if ok && reg.cp != nil {
|
||||
cp := reg.cp
|
||||
cs.log.INFO.Printf("charge point connected, registering: %s", chargePoint.ID())
|
||||
|
||||
// update id
|
||||
cp.RegisterID(chargePoint.ID())
|
||||
|
||||
cs.cps[chargePoint.ID()] = cp
|
||||
delete(cs.cps, "")
|
||||
cs.regs[chargePoint.ID()].cp = cp
|
||||
delete(cs.regs, "")
|
||||
|
||||
cp.connect(true)
|
||||
|
||||
|
|
@ -133,7 +142,7 @@ func (cs *CS) NewChargePoint(chargePoint ocpp16.ChargePointConnection) {
|
|||
|
||||
// register unknown charge point
|
||||
// when charge point setup is complete, it will eventually be associated with the connected id
|
||||
cs.cps[chargePoint.ID()] = nil
|
||||
cs.regs[chargePoint.ID()] = new(registration)
|
||||
}
|
||||
|
||||
// ChargePointDisconnected implements ocpp16.ChargePointConnectionHandler
|
||||
|
|
|
|||
|
|
@ -66,6 +66,16 @@ func (cs *CS) OnStatusNotification(id string, request *core.StatusNotificationRe
|
|||
return cp.OnStatusNotification(request)
|
||||
}
|
||||
|
||||
cs.mu.Lock()
|
||||
defer cs.mu.Unlock()
|
||||
|
||||
// cache status for future cp connection
|
||||
if reg, ok := cs.regs[id]; ok {
|
||||
reg.mu.Lock()
|
||||
reg.status = request
|
||||
reg.mu.Unlock()
|
||||
}
|
||||
|
||||
return new(core.StatusNotificationConfirmation), nil
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -40,8 +40,7 @@ func Instance() *CS {
|
|||
|
||||
instance = &CS{
|
||||
log: log,
|
||||
cps: make(map[string]*CP),
|
||||
init: make(map[string]*sync.Mutex),
|
||||
regs: make(map[string]*registration),
|
||||
CentralSystem: cs,
|
||||
}
|
||||
instance.txnId.Store(time.Now().UTC().Unix())
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue