Include new callback and context logic in v1.6 and v2.0.1 charging stations

Signed-off-by: Lorenzo <[email protected]>

Author: lorenzodonini

ocpp1.6/charge_point.go

@@ -1,10 +1,10 @@
 package ocpp16
 
 import (
+	"context"
 	"fmt"
 	"reflect"
 
-	"github.com/lorenzodonini/ocpp-go/internal/callbackqueue"
 	"github.com/lorenzodonini/ocpp-go/ocpp"
 	"github.com/lorenzodonini/ocpp-go/ocpp1.6/certificates"
 	"github.com/lorenzodonini/ocpp-go/ocpp1.6/core"
@@ -47,10 +47,15 @@ func (cp *chargePoint) error(err error) {
 	}
 }
 
-// Callback invoked whenever a queued request is canceled, due to timeout.
-// By default, the callback returns a GenericError to the caller, who sent the original request.
-func (cp *chargePoint) onRequestTimeout(_ string, _ ocpp.Request, err *ocpp.Error) {
-	cp.errorHandler <- err
+// Function invoked whenever a Response or an error is received for a request.
+// The callback includes the user-defined callback function, which may be used to handle the response accordingly.
+// The identifier of the original request is attached to the function as reference.
+func (cp *chargePoint) onResponse(requestID string, response ocpp.Response, err error, callback ocpp.Callback) {
+	if callback == nil {
+		cp.error(fmt.Errorf("no callback defined for confirmation to request %v, dropping confirmation", requestID))
+		return
+	}
+	callback(response, err)
 }
 
 // Errors returns a channel for error messages. If it doesn't exist it es created.
@@ -284,39 +289,54 @@ func (cp *chargePoint) SetCertificateHandler(handler certificates.ChargePointHan
 }
 
 func (cp *chargePoint) SendRequest(request ocpp.Request) (ocpp.Response, error) {
+	return cp.SendRequestWithContext(request, context.TODO())
+}
+
+func (cp *chargePoint) SendRequestWithContext(request ocpp.Request, ctx context.Context) (ocpp.Response, error) {
 	featureName := request.GetFeatureName()
 	if _, found := cp.client.GetProfileForFeature(featureName); !found {
 		return nil, fmt.Errorf("feature %v is unsupported on charge point (missing profile), cannot send request", featureName)
 	}
-
+	if ctx == nil {
+		ctx = context.TODO()
+	}
 	// Wraps an asynchronous response
 	type asyncResponse struct {
 		r ocpp.Response
 		e error
 	}
-	// Create channel and pass it to a callback function, for retrieving asynchronous response
+	// Create channel and use it within a callback function, for retrieving asynchronous response
 	asyncResponseC := make(chan asyncResponse, 1)
-	send := func() error {
-		return cp.client.SendRequest(request)
+	callback := func(response ocpp.Response, err error) {
+		if ctx.Err() != nil {
+			// Request was canceled already, ignore callback.
+			// Confirmation will be handled by select switch below.
+			return
+		}
+		asyncResponseC <- asyncResponse{response, err}
 	}
-	err := cp.callbacks.TryQueue("main", send, func(confirmation ocpp.Response, err error) {
-		asyncResponseC <- asyncResponse{r: confirmation, e: err}
-	})
+	// Send request, then start blocking wait for asynchronous response
+	_, err := cp.client.SendRequest(request, callback, ctx)
 	if err != nil {
 		return nil, err
 	}
 	select {
 	case asyncResult, ok := <-asyncResponseC:
 		if !ok {
-			return nil, fmt.Errorf("internal error while receiving result for %v request", request.GetFeatureName())
+			return nil, fmt.Errorf("internal error while receiving result for %s request", request.GetFeatureName())
 		}
 		return asyncResult.r, asyncResult.e
-	case <-cp.stopC:
-		return nil, fmt.Errorf("client stopped while waiting for response to %v", request.GetFeatureName())
+	case <-ctx.Done():
+		// Request was canceled by user. Return an error and don't handle callbacks any longer
+		return nil, fmt.Errorf("request timed out/canceled by user. Any incoming response will be ignored")
 	}
 }
 
-func (cp *chargePoint) SendRequestAsync(request ocpp.Request, callback func(confirmation ocpp.Response, err error)) error {
+func (cp *chargePoint) SendRequestAsync(request ocpp.Request, callback ocpp.Callback) error {
+	return cp.SendRequestAsyncWithContext(request, callback, context.TODO())
+}
+
+func (cp *chargePoint) SendRequestAsyncWithContext(request ocpp.Request, callback ocpp.Callback, ctx context.Context) error {
 	featureName := request.GetFeatureName()
 	if _, found := cp.client.GetProfileForFeature(featureName); !found {
 		return fmt.Errorf("feature %v is unsupported on charge point (missing profile), cannot send request", featureName)
@@ -331,51 +351,14 @@ func (cp *chargePoint) SendRequestAsync(request ocpp.Request, callback func(conf
 	default:
 		return fmt.Errorf("unsupported action %v on charge point, cannot send request", featureName)
 	}
-	// Response will be retrieved asynchronously via asyncHandler
-	send := func() error {
-		return cp.client.SendRequest(request)
+	if ctx == nil {
+		ctx = context.TODO()
 	}
-	err := cp.callbacks.TryQueue("main", send, callback)
+	// Response will be handled asynchronously and passed to the callback function
+	_, err := cp.client.SendRequest(request, callback, ctx)
 	return err
 }
 
-func (cp *chargePoint) asyncCallbackHandler() {
-	for {
-		select {
-		case confirmation := <-cp.confirmationHandler:
-			// Get and invoke callback
-			if callback, ok := cp.callbacks.Dequeue("main"); ok {
-				callback(confirmation, nil)
-			} else {
-				err := fmt.Errorf("no handler available for incoming response %v", confirmation.GetFeatureName())
-				cp.error(err)
-			}
-		case protoError := <-cp.errorHandler:
-			// Get and invoke callback
-			if callback, ok := cp.callbacks.Dequeue("main"); ok {
-				callback(nil, protoError)
-			} else {
-				err := fmt.Errorf("no handler available for error %v", protoError.Error())
-				cp.error(err)
-			}
-		case <-cp.stopC:
-			// Handler stopped, cleanup callbacks.
-			// No callback invocation, since the user manually stopped the client.
-			cp.clearCallbacks(false)
-			return
-		}
-	}
-}
-
-func (cp *chargePoint) clearCallbacks(invokeCallback bool) {
-	for cb, ok := cp.callbacks.Dequeue("main"); ok; cb, ok = cp.callbacks.Dequeue("main") {
-		if invokeCallback {
-			err := ocpp.NewError(ocppj.GenericError, "client stopped, no response received from server", "")
-			cb(nil, err)
-		}
-	}
-}
-
 func (cp *chargePoint) sendResponse(confirmation ocpp.Response, err error, requestId string) {
 	if err != nil {
 		// Send error response

ocpp1.6/v16.go

@@ -2,10 +2,10 @@
 package ocpp16
 
 import (
+	"context"
 	"crypto/tls"
 	"net"
 
-	"github.com/lorenzodonini/ocpp-go/internal/callbackqueue"
 	"github.com/lorenzodonini/ocpp-go/ocpp"
 	"github.com/lorenzodonini/ocpp-go/ocpp1.6/certificates"
 	"github.com/lorenzodonini/ocpp-go/ocpp1.6/core"
@@ -110,11 +110,30 @@ type ChargePoint interface {
 	//
 	// The request is synchronous blocking.
 	SendRequest(request ocpp.Request) (ocpp.Response, error)
+	// Sends a request to the central system.
+	// The central system will respond with a confirmation, or with an error if the request was invalid or could not be processed.
+	// In case of network issues (i.e. the remote host couldn't be reached), the function also returns an error.
+	//
+	// An optional context can be passed to the function. If the context is canceled before the request is completed,
+	// the function returns an error.
+	//
+	// The request is synchronous blocking.
+	SendRequestWithContext(request ocpp.Request, ctx context.Context) (ocpp.Response, error)
 	// Sends an asynchronous request to the central system.
-	// The central system will respond with a confirmation messages, or with an error if the request was invalid or could not be processed.
+	// The central system will reply with a confirmation message, or with an error if the request was invalid or could not be processed.
 	// This result is propagated via a callback, called asynchronously.
+	//
+	// In case of network issues (i.e. the remote host couldn't be reached), the function returns an error directly. In this case, the callback is never called.
+	SendRequestAsync(request ocpp.Request, callback ocpp.Callback) error
+	// Sends an asynchronous request to the central system.
+	// The central system will reply with a confirmation message, or with an error if the request was invalid or could not be processed.
+	// This result is propagated via a callback, called asynchronously.
+	//
 	// In case of network issues (i.e. the remote host couldn't be reached), the function returns an error directly. In this case, the callback is never called.
-	SendRequestAsync(request ocpp.Request, callback func(confirmation ocpp.Response, protoError error)) error
+	//
+	// An optional context can be passed to the function. If the context is canceled before the request is completed,
+	// the callback is invoked with an error.
+	SendRequestAsyncWithContext(request ocpp.Request, callback ocpp.Callback, ctx context.Context) error
 	// Connects to the central system and starts the charge point routine.
 	// The function doesn't block and returns right away, after having attempted to open a connection to the central system.
 	// If the connection couldn't be opened, an error is returned.
@@ -204,6 +223,12 @@ func NewChargePoint(id string, endpoint *ocppj.Client, client ws.Client) ChargeP
 		cp.errorHandler <- err
 	})
 	cp.client.SetRequestHandler(cp.handleIncomingRequest)
+	// Callback invoked by ocppj layer, whenever:
+	//  - a response/error is received
+	// 	- an internal error occurs
+	// 	- the request is cancelled
+	//  - the request times out
+	cp.client.SetResponseHandler(cp.onResponse)
 	return &cp
 }
 

ocpp2.0.1/charging_station.go

@@ -1,10 +1,10 @@
 package ocpp2
 
 import (
+	"context"
 	"fmt"
 	"reflect"
 
-	"github.com/lorenzodonini/ocpp-go/internal/callbackqueue"
 	"github.com/lorenzodonini/ocpp-go/ocpp"
 	"github.com/lorenzodonini/ocpp-go/ocpp2.0.1/authorization"
 	"github.com/lorenzodonini/ocpp-go/ocpp2.0.1/availability"
@@ -44,9 +44,6 @@ type chargingStation struct {
 	diagnosticsHandler   diagnostics.ChargingStationHandler
 	displayHandler       display.ChargingStationHandler
 	dataHandler          data.ChargingStationHandler
-	responseHandler      chan ocpp.Response
-	errorHandler         chan error
-	callbacks            callbackqueue.CallbackQueue
 	stopC                chan struct{}
 	errC                 chan error // external error channel
 }
@@ -65,10 +62,15 @@ func (cs *chargingStation) Errors() <-chan error {
 	return cs.errC
 }
 
-// Callback invoked whenever a queued request is canceled, due to timeout.
-// By default, the callback returns a GenericError to the caller, who sent the original request.
-func (cs *chargingStation) onRequestTimeout(_ string, _ ocpp.Request, err *ocpp.Error) {
-	cs.errorHandler <- err
+// Function invoked whenever a Response or an error is received for a request.
+// The callback includes the user-defined callback function, which may be used to handle the response accordingly.
+// A reference to the original request is attached to the function, so that the user may identify the request.
+func (cs *chargingStation) onResponse(requestID string, response ocpp.Response, err error, callback ocpp.Callback) {
+	if callback == nil {
+		cs.error(fmt.Errorf("no callback defined for response to request %v, dropping response", requestID))
+		return
+	}
+	callback(response, err)
 }
 
 func (cs *chargingStation) BootNotification(reason provisioning.BootReason, model string, vendor string, props ...func(request *provisioning.BootNotificationRequest)) (*provisioning.BootNotificationResponse, error) {
@@ -461,10 +463,17 @@ func (cs *chargingStation) SetDataHandler(handler data.ChargingStationHandler) {
 }
 
 func (cs *chargingStation) SendRequest(request ocpp.Request) (ocpp.Response, error) {
+	return cs.SendRequestWithContext(request, context.TODO())
+}
+
+func (cs *chargingStation) SendRequestWithContext(request ocpp.Request, ctx context.Context) (ocpp.Response, error) {
 	featureName := request.GetFeatureName()
 	if _, found := cs.client.GetProfileForFeature(featureName); !found {
 		return nil, fmt.Errorf("feature %v is unsupported on charging station (missing profile), cannot send request", featureName)
 	}
+	if ctx == nil {
+		ctx = context.TODO()
+	}
 
 	// Wraps an asynchronous response
 	type asyncResponse struct {
@@ -473,23 +482,36 @@ func (cs *chargingStation) SendRequest(request ocpp.Request) (ocpp.Response, err
 	}
 	// Create channel and pass it to a callback function, for retrieving asynchronous response
 	asyncResponseC := make(chan asyncResponse, 1)
-	send := func() error {
-		return cs.client.SendRequest(request)
+	callback := func(response ocpp.Response, err error) {
+		if ctx.Err() != nil {
+			// Request was canceled already, ignore callback.
+			// Response will be handled by select statement below.
+			return
+		}
+		asyncResponseC <- asyncResponse{response, err}
 	}
-	err := cs.callbacks.TryQueue("main", send, func(confirmation ocpp.Response, err error) {
-		asyncResponseC <- asyncResponse{r: confirmation, e: err}
-	})
+	// Send request, then start blocking wait for asynchronous response
+	_, err := cs.client.SendRequest(request, callback, ctx)
 	if err != nil {
 		return nil, err
 	}
-	asyncResult, ok := <-asyncResponseC
-	if !ok {
-		return nil, fmt.Errorf("internal error while receiving result for %v request", request.GetFeatureName())
+	select {
+	case asyncResult, ok := <-asyncResponseC:
+		if !ok {
+			return nil, fmt.Errorf("internal error while receiving result for %s request", request.GetFeatureName())
+		}
+		return asyncResult.r, asyncResult.e
+	case <-ctx.Done():
+		// Request was canceled by user. Return an error and don't handle callbacks any longer
+		return nil, fmt.Errorf("request timed out/canceled by user. Any incoming response will be ignored")
 	}
-	return asyncResult.r, asyncResult.e
 }
 
-func (cs *chargingStation) SendRequestAsync(request ocpp.Request, callback func(response ocpp.Response, err error)) error {
+func (cs *chargingStation) SendRequestAsync(request ocpp.Request, callback ocpp.Callback) error {
+	return cs.SendRequestAsyncWithContext(request, callback, context.TODO())
+}
+
+func (cs *chargingStation) SendRequestAsyncWithContext(request ocpp.Request, callback ocpp.Callback, ctx context.Context) error {
 	featureName := request.GetFeatureName()
 	if _, found := cs.client.GetProfileForFeature(featureName); !found {
 		return fmt.Errorf("feature %v is unsupported on charging station (missing profile), cannot send request", featureName)
@@ -524,37 +546,14 @@ func (cs *chargingStation) SendRequestAsync(request ocpp.Request, callback func(
 	default:
 		return fmt.Errorf("unsupported action %v on charging station, cannot send request", featureName)
 	}
-	// Response will be retrieved asynchronously via asyncHandler
-	send := func() error {
-		return cs.client.SendRequest(request)
+	if ctx == nil {
+		ctx = context.TODO()
 	}
-	err := cs.callbacks.TryQueue("main", send, callback)
+	// Response will be handled asynchronously and passed to the callback function
+	_, err := cs.client.SendRequest(request, callback, ctx)
 	return err
 }
 
-func (cs *chargingStation) asyncCallbackHandler() {
-	for {
-		select {
-		case confirmation := <-cs.responseHandler:
-			// Get and invoke callback
-			if callback, ok := cs.callbacks.Dequeue("main"); ok {
-				callback(confirmation, nil)
-			} else {
-				cs.error(fmt.Errorf("no callback available for incoming response %v", confirmation.GetFeatureName()))
-			}
-		case protoError := <-cs.errorHandler:
-			// Get and invoke callback
-			if callback, ok := cs.callbacks.Dequeue("main"); ok {
-				callback(nil, protoError)
-			} else {
-				cs.error(fmt.Errorf("no callback available for incoming error %w", protoError))
-			}
-		case <-cs.stopC:
-			return
-		}
-	}
-}
-
 func (cs *chargingStation) sendResponse(response ocpp.Response, err error, requestId string) {
 	if err != nil {
 		// Send error response