feat: added support for send and call result err events

Author: xBlaz3kx

ocppj/client.go

@@ -3,6 +3,7 @@ package ocppj
 import (
 	"errors"
 	"fmt"
+	"strings"
 
 	"gopkg.in/go-playground/validator.v9"
 
@@ -208,6 +209,35 @@ func (c *Client) SendRequest(request ocpp.Request) error {
 	return nil
 }
 
+func (c *Client) SendEvent(request ocpp.Request) error {
+	if !c.dispatcher.IsRunning() {
+		return fmt.Errorf("ocppj client is not started, couldn't send request")
+	}
+
+	// Check if the request feature is a Stream feature
+	feature := request.GetFeatureName()
+	if !strings.HasSuffix(feature, "Stream") {
+		return fmt.Errorf("ocppj client can only send events for Stream features, got: %s", feature)
+	}
+
+	call, err := c.CreateCall(request)
+	if err != nil {
+		return err
+	}
+	jsonMessage, err := call.MarshalJSON()
+	if err != nil {
+		return err
+	}
+	// Message will be processed by dispatcher. A dedicated mechanism allows to delegate the message queue handling.
+	if err = c.dispatcher.SendRequest(RequestBundle{Call: call, Data: jsonMessage}); err != nil {
+		log.Errorf("error dispatching request [%s, %s]: %v", call.UniqueId, call.Action, err)
+		return err
+	}
+
+	log.Debugf("enqueued CALL [%s, %s]", call.UniqueId, call.Action)
+	return nil
+}
+
 // Sends an OCPP Response to the server.
 // The requestID parameter is required and identifies the previously received request.
 //
@@ -313,6 +343,12 @@ func (c *Client) ocppMessageHandler(data []byte) error {
 			if c.errorHandler != nil {
 				c.errorHandler(ocpp.NewError(callError.ErrorCode, callError.ErrorDescription, callError.UniqueId), callError.ErrorDetails)
 			}
+		case CALL_RESULT_ERROR:
+			callError := message.(*CallResultError)
+			log.Debugf("handling incoming CALL RESULT ERROR [%s]", callError.UniqueId)
+			if c.errorHandler != nil {
+				c.errorHandler(ocpp.NewError(callError.ErrorCode, callError.ErrorDescription, callError.UniqueId), callError.ErrorDetails)
+			}
 		}
 	}
 	return nil

ocppj/ocppj.go

@@ -182,7 +182,41 @@ func (callError *CallError) MarshalJSON() ([]byte, error) {
 	return ocppMessageToJson(fields)
 }
 
-// -------------------- Call --------------------
+// -------------------- Call Result Error --------------------
+
+// An OCPP-J CallResultError message, containing an OCPP Result Error.
+type CallResultError struct {
+	Message
+	MessageTypeId    MessageType    `json:"messageTypeId" validate:"required,eq=4"`
+	UniqueId         string         `json:"uniqueId" validate:"required,max=36"`
+	ErrorCode        ocpp.ErrorCode `json:"errorCode" validate:"errorCode"`
+	ErrorDescription string         `json:"errorDescription" validate:"omitempty,max=255"`
+	ErrorDetails     interface{}    `json:"errorDetails" validate:"omitempty"`
+}
+
+func (callError *CallResultError) GetMessageTypeId() MessageType {
+	return callError.MessageTypeId
+}
+
+func (callError *CallResultError) GetUniqueId() string {
+	return callError.UniqueId
+}
+
+func (callError *CallResultError) MarshalJSON() ([]byte, error) {
+	fields := make([]interface{}, 5)
+	fields[0] = int(callError.MessageTypeId)
+	fields[1] = callError.UniqueId
+	fields[2] = callError.ErrorCode
+	fields[3] = callError.ErrorDescription
+	if callError.ErrorDetails == nil {
+		fields[4] = struct{}{}
+	} else {
+		fields[4] = callError.ErrorDetails
+	}
+	return ocppMessageToJson(fields)
+}
+
+// -------------------- Send --------------------
 
 // An OCPP-J SEND message, containing an OCPP Request.
 type Send struct {
@@ -259,7 +293,7 @@ func IsErrorCodeValid(fl validator.FieldLevel) bool {
 	case NotImplemented, NotSupported, InternalError, MessageTypeNotSupported,
 		ProtocolError, SecurityError, FormatViolationV16, FormatViolationV2,
 		PropertyConstraintViolation, OccurrenceConstraintViolationV16, OccurrenceConstraintViolationV2,
-		TypeConstraintViolation, GenericError:
+		TypeConstraintViolation, GenericError, RpcFrameworkError:
 		return true
 	}
 	return false
@@ -526,6 +560,47 @@ func (endpoint *Endpoint) CreateCall(request ocpp.Request) (*Call, error) {
 	return &call, nil
 }
 
+func (endpoint *Endpoint) CreateSend(request ocpp.Request) (*Send, error) {
+	action := request.GetFeatureName()
+	profile, _ := endpoint.GetProfileForFeature(action)
+	if profile == nil {
+		return nil, fmt.Errorf("Couldn't create Send for unsupported action %v", action)
+	}
+	// TODO: handle collisions?
+	uniqueId := messageIdGenerator()
+	call := Send{
+		MessageTypeId: SEND,
+		UniqueId:      uniqueId,
+		Action:        action,
+		Payload:       request,
+	}
+	if validationEnabled {
+		err := Validate.Struct(call)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return &call, nil
+}
+
+// Creates a CallResultError message, given the message's unique ID and the error.
+func (endpoint *Endpoint) CreateCallResultError(uniqueId string, code ocpp.ErrorCode, description string, details interface{}) (*CallResultError, error) {
+	callError := CallResultError{
+		MessageTypeId:    CALL_RESULT_ERROR,
+		UniqueId:         uniqueId,
+		ErrorCode:        code,
+		ErrorDescription: description,
+		ErrorDetails:     details,
+	}
+	if validationEnabled {
+		err := Validate.Struct(callError)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return &callError, nil
+}
+
 // Creates a CallResult message, given an OCPP response and the message's unique ID.
 //
 // Returns an error in case the response's feature is not supported on this endpoint.

ocppj/ocppj_test.go

@@ -292,6 +292,15 @@ func CheckCall(call *ocppj.Call, t *testing.T, expectedAction string, expectedId
 	assert.Nil(t, err)
 }
 
+func CheckSend(call *ocppj.Send, t *testing.T, expectedAction string, expectedId string) {
+	assert.Equal(t, ocppj.SEND, call.GetMessageTypeId())
+	assert.Equal(t, expectedAction, call.Action)
+	assert.Equal(t, expectedId, call.GetUniqueId())
+	assert.NotNil(t, call.Payload)
+	err := Validate.Struct(call)
+	assert.Nil(t, err)
+}
+
 func ParseCallResult(endpoint *ocppj.Endpoint, state ocppj.ClientState, json string, t *testing.T) *ocppj.CallResult {
 	parsedData, err := ocppj.ParseJsonMessage(json)
 	require.NoError(t, err)
@@ -336,6 +345,16 @@ func CheckCallError(t *testing.T, callError *ocppj.CallError, expectedId string,
 	assert.Nil(t, err)
 }
 
+func CheckCallResultError(t *testing.T, callError *ocppj.CallResultError, expectedId string, expectedError ocpp.ErrorCode, expectedDescription string, expectedDetails interface{}) {
+	assert.Equal(t, ocppj.CALL_RESULT_ERROR, callError.GetMessageTypeId())
+	assert.Equal(t, expectedId, callError.GetUniqueId())
+	assert.Equal(t, expectedError, callError.ErrorCode)
+	assert.Equal(t, expectedDescription, callError.ErrorDescription)
+	assert.Equal(t, expectedDetails, callError.ErrorDetails)
+	err := Validate.Struct(callError)
+	assert.Nil(t, err)
+}
+
 func assertPanic(t *testing.T, f func(), recoveredAssertion func(interface{})) {
 	defer func() {
 		r := recover()
@@ -496,6 +515,23 @@ func (suite *OcppJTestSuite) TestCreateCall() {
 	assert.Nil(t, pendingRequest)
 }
 
+func (suite *OcppJTestSuite) TestCreateSend() {
+	t := suite.T()
+	mockValue := "somevalue"
+	request := newMockRequest(mockValue)
+	call, err := suite.chargePoint.CreateSend(request)
+	assert.Nil(t, err)
+	CheckSend(call, t, MockFeatureName+"Stream", call.UniqueId)
+	message, ok := call.Payload.(*MockRequest)
+	assert.True(t, ok)
+	assert.NotNil(t, message)
+	assert.Equal(t, mockValue, message.MockValue)
+	// Check that request was not yet stored as pending request
+	pendingRequest, exists := suite.chargePoint.RequestState.GetPendingRequest(call.UniqueId)
+	assert.False(t, exists)
+	assert.Nil(t, pendingRequest)
+}
+
 func (suite *OcppJTestSuite) TestCreateCallResult() {
 	t := suite.T()
 	mockValue := "someothervalue"
@@ -525,6 +561,21 @@ func (suite *OcppJTestSuite) TestCreateCallError() {
 	CheckCallError(t, callError, mockUniqueId, ocppj.GenericError, mockDescription, mockDetails)
 }
 
+func (suite *OcppJTestSuite) TestCreateCallResultError() {
+	t := suite.T()
+	mockUniqueId := "123456"
+	mockDescription := "somedescription"
+	mockDetailString := "somedetailstring"
+	type MockDetails struct {
+		DetailString string
+	}
+	mockDetails := MockDetails{DetailString: mockDetailString}
+	callError, err := suite.chargePoint.CreateCallResultError(mockUniqueId, ocppj.GenericError, mockDescription, mockDetails)
+	assert.Nil(t, err)
+	assert.NotNil(t, callError)
+	CheckCallResultError(t, callError, mockUniqueId, ocppj.GenericError, mockDescription, mockDetails)
+}
+
 func (suite *OcppJTestSuite) TestParseMessageInvalidLength() {
 	t := suite.T()
 	mockMessage := make([]interface{}, 2)