Merge pull request #10 from satimoto/release/v0.4.0

v0.4.0

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/joho/godotenv v1.4.0
 	github.com/nsf/jsondiff v0.0.0-20210926074059-1e845ec5d249
 	github.com/prometheus/client_golang v1.13.1
-	github.com/satimoto/go-datastore v0.3.1-0.20221215215552-467363aef4a9
+	github.com/satimoto/go-datastore v0.3.1-0.20230102194740-a5839718733c
 	google.golang.org/grpc v1.47.0
 )
 

go.sum

@@ -402,6 +402,8 @@ github.com/satimoto/go-datastore v0.3.1-0.20221211215042-d5318660b0e1 h1:j8SMRJQ
 github.com/satimoto/go-datastore v0.3.1-0.20221211215042-d5318660b0e1/go.mod h1:SvM8losYPwH6hJgUKLkjNtQwVv06lZgCMWl8cGdllCM=
 github.com/satimoto/go-datastore v0.3.1-0.20221215215552-467363aef4a9 h1:vmWEl6/uFLQND2y0IE47Ym8iBDRhfF/z8AdFH9ls8LI=
 github.com/satimoto/go-datastore v0.3.1-0.20221215215552-467363aef4a9/go.mod h1:SvM8losYPwH6hJgUKLkjNtQwVv06lZgCMWl8cGdllCM=
+github.com/satimoto/go-datastore v0.3.1-0.20230102194740-a5839718733c h1:pFctByHq4fH9RKHDdHTcyfFRspAI6WJUuBZNVMkzZGg=
+github.com/satimoto/go-datastore v0.3.1-0.20230102194740-a5839718733c/go.mod h1:SvM8losYPwH6hJgUKLkjNtQwVv06lZgCMWl8cGdllCM=
 github.com/satimoto/go-lsp v0.2.1-0.20221115190646-b981e98c7ba4 h1:3QKFoeOVSDU8bvrVfj/zAMDq+hg3KJBLaMgEgutJExk=
 github.com/satimoto/go-lsp v0.2.1-0.20221115190646-b981e98c7ba4/go.mod h1:bleUpVumUC8QAFi0CZdnbtDRGWRFKETMH5JWcLAolBs=
 github.com/sergi/go-diff v1.1.0 h1:we8PVUC3FE2uYfodKH/nBHMSetSfHDR6scGdBi+erh0=

internal/async/service.go

@@ -1,5 +1,7 @@
 package async
 
+import "log"
+
 type AsyncResult struct {
 	String string
 	Bool bool
@@ -16,11 +18,13 @@ func NewService() *AsyncService {
 }
 
 func (r *AsyncService) Add(key string) <-chan AsyncResult {
+	log.Printf("Async: Add key=%v", key)
 	r.channels[key] = make(chan AsyncResult)
 	return r.channels[key]
 }
 
 func (r *AsyncService) Remove(key string) {
+	log.Printf("Async: Remove key=%v", key)
 	if _, ok := r.channels[key]; ok {
 		close(r.channels[key])
 		delete(r.channels, key)
@@ -29,9 +33,11 @@ func (r *AsyncService) Remove(key string) {
 
 func (r *AsyncService) Set(key string, result AsyncResult) bool {
 	if _, ok := r.channels[key]; ok {
+		log.Printf("Async: Set key=%v", key)
 		r.channels[key] <-result
 		return true
 	}
 
+	log.Printf("Async: Invalid key=%v", key)
 	return false
 }

internal/cdr/v2.1.1/sync.go

@@ -11,6 +11,7 @@ import (
 	"github.com/satimoto/go-datastore/pkg/db"
 	"github.com/satimoto/go-datastore/pkg/util"
 	coreCdr "github.com/satimoto/go-ocpi/internal/cdr"
+	dto "github.com/satimoto/go-ocpi/internal/dto/v2.1.1"
 	metrics "github.com/satimoto/go-ocpi/internal/metric"
 	"github.com/satimoto/go-ocpi/internal/transportation"
 )
@@ -59,42 +60,21 @@ func (r *CdrResolver) SyncByIdentifier(ctx context.Context, credential db.Creden
 	}
 
 	for {
+		if retries >= 5 {
+			break
+		}
+
 		query.Set("limit", fmt.Sprintf("%d", limit))
 		query.Set("offset", fmt.Sprintf("%d", offset))
 		requestUrl.RawQuery = query.Encode()
 
-		response, err := r.OcpiService.Do(http.MethodGet, requestUrl.String(), header, nil)
+		dto, limit := r.sendRequest(requestUrl.String(), header, limit)
 
-		if err != nil {
-			metrics.RecordError("OCPI030", "Error making request", err)
-			log.Printf("OCPI030: Method=%v, Url=%v, Header=%#v", http.MethodGet, requestUrl.String(), header)
+		if dto == nil {
 			retries++
-
-			if retries >= 5 {
-				break
-			}
-
 			continue
 		}
 
-		dto, err := r.UnmarshalPullDto(response.Body)
-		defer response.Body.Close()
-
-		if err != nil {
-			metrics.RecordError("OCPI031", "Error unmarshaling response", err)
-			util.LogHttpResponse("OCPI031", requestUrl.String(), response, true)
-			break
-		}
-
-		limit = transportation.GetXLimitHeader(response, limit)
-
-		if dto.StatusCode != transportation.STATUS_CODE_OK {
-			metrics.RecordError("OCPI032", "Error response failure", err)
-			util.LogHttpRequest("OCPI032", requestUrl.String(), response.Request, true)
-			util.LogHttpResponse("OCPI032", requestUrl.String(), response, true)
-			break
-		}
-
 		retries = 0
 
 		if dto.StatusCode == transportation.STATUS_CODE_OK {
@@ -110,3 +90,35 @@ func (r *CdrResolver) SyncByIdentifier(ctx context.Context, credential db.Creden
 		}
 	}
 }
+
+func (r *CdrResolver) sendRequest(url string, header transportation.OcpiRequestHeader, limit int) (*dto.OcpiCdrsDto, int) {
+	response, err := r.OcpiService.Do(http.MethodGet, url, header, nil)
+
+	if err != nil {
+		metrics.RecordError("OCPI030", "Error making request", err)
+		log.Printf("OCPI030: Method=%v, Url=%v, Header=%#v", http.MethodGet, url, header)
+
+		return nil, limit
+	}
+
+	dto, err := r.UnmarshalPullDto(response.Body)
+	defer response.Body.Close()
+
+	if err != nil {
+		metrics.RecordError("OCPI031", "Error unmarshaling response", err)
+		util.LogHttpResponse("OCPI031", url, response, true)
+
+		return nil, limit
+	}
+
+	if dto.StatusCode != transportation.STATUS_CODE_OK {
+		metrics.RecordError("OCPI032", "Error response failure", err)
+		util.LogHttpRequest("OCPI032", url, response.Request, true)
+		util.LogHttpResponse("OCPI032", url, response, true)
+		log.Printf("OCPI032: StatusCode=%v, StatusMessage=%v", dto.StatusCode, dto.StatusMessage)
+
+		return nil, limit
+	}
+
+	return dto, transportation.GetXLimitHeader(response, limit)
+}

internal/location/v2.1.1/sync.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/satimoto/go-datastore/pkg/db"
 	"github.com/satimoto/go-datastore/pkg/util"
+	dto "github.com/satimoto/go-ocpi/internal/dto/v2.1.1"
 	coreLocation "github.com/satimoto/go-ocpi/internal/location"
 	metrics "github.com/satimoto/go-ocpi/internal/metric"
 	"github.com/satimoto/go-ocpi/internal/transportation"
@@ -64,43 +65,21 @@ func (r *LocationResolver) SyncByIdentifier(ctx context.Context, credential db.C
 	}
 
 	for {
+		if retries >= 5 {
+			break
+		}
+
 		query.Set("limit", fmt.Sprintf("%d", limit))
 		query.Set("offset", fmt.Sprintf("%d", offset))
 		requestUrl.RawQuery = query.Encode()
 
-		response, err := r.OcpiService.Do(http.MethodGet, requestUrl.String(), header, nil)
+		dto, limit := r.sendRequest(requestUrl.String(), header, limit)
 
-		if err != nil {
-			metrics.RecordError("OCPI127", "Error making request", err)
-			log.Printf("OCPI127: Method=%v, Url=%v, Header=%#v", http.MethodGet, requestUrl.String(), header)
+		if dto == nil {
 			retries++
-
-			if retries >= 5 {
-				break
-			}
-
 			continue
 		}
 
-		dto, err := r.UnmarshalPullDto(response.Body)
-		defer response.Body.Close()
-
-		if err != nil {
-			metrics.RecordError("OCPI128", "Error unmarshaling response", err)
-			util.LogHttpResponse("OCPI128", requestUrl.String(), response, true)
-			break
-		}
-
-		limit = transportation.GetXLimitHeader(response, limit)
-
-		if dto.StatusCode != transportation.STATUS_CODE_OK {
-			metrics.RecordError("OCPI129", "Error response failure", err)
-			util.LogHttpRequest("OCPI129", versionEndpoint.Url, response.Request, true)
-			util.LogHttpResponse("OCPI129", requestUrl.String(), response, true)
-			log.Printf("OCPI129: StatusCode=%v, StatusMessage=%v", dto.StatusCode, dto.StatusMessage)
-			break
-		}
-
 		retries = 0
 
 		if dto.StatusCode == transportation.STATUS_CODE_OK {
@@ -116,3 +95,35 @@ func (r *LocationResolver) SyncByIdentifier(ctx context.Context, credential db.C
 		}
 	}
 }
+
+func (r *LocationResolver) sendRequest(url string, header transportation.OcpiRequestHeader, limit int) (*dto.OcpiLocationsDto, int) {
+	response, err := r.OcpiService.Do(http.MethodGet, url, header, nil)
+
+	if err != nil {
+		metrics.RecordError("OCPI127", "Error making request", err)
+		log.Printf("OCPI127: Method=%v, Url=%v, Header=%#v", http.MethodGet, url, header)
+
+		return nil, limit
+	}
+
+	dto, err := r.UnmarshalPullDto(response.Body)
+	defer response.Body.Close()
+
+	if err != nil {
+		metrics.RecordError("OCPI128", "Error unmarshaling response", err)
+		util.LogHttpResponse("OCPI128", url, response, true)
+
+		return nil, limit
+	}
+
+	if dto.StatusCode != transportation.STATUS_CODE_OK {
+		metrics.RecordError("OCPI129", "Error response failure", err)
+		util.LogHttpRequest("OCPI129", url, response.Request, true)
+		util.LogHttpResponse("OCPI129", url, response, true)
+		log.Printf("OCPI129: StatusCode=%v, StatusMessage=%v", dto.StatusCode, dto.StatusMessage)
+
+		return nil, limit
+	}
+
+	return dto, transportation.GetXLimitHeader(response, limit)
+}

internal/rpc/command/rpc.go

@@ -12,7 +12,6 @@ import (
 	metrics "github.com/satimoto/go-ocpi/internal/metric"
 	"github.com/satimoto/go-ocpi/ocpirpc"
 	ocpiCommand "github.com/satimoto/go-ocpi/pkg/ocpi/command"
-	ocpiTokenAuthorization "github.com/satimoto/go-ocpi/pkg/ocpi/tokenauthorization"
 )
 
 func (r *RpcCommandResolver) ReserveNow(reqCtx context.Context, input *ocpirpc.ReserveNowRequest) (*ocpirpc.ReserveNowResponse, error) {
@@ -132,14 +131,6 @@ func (r *RpcCommandResolver) StartSession(reqCtx context.Context, input *ocpirpc
 			return nil, errors.New("error starting session")
 		}
 
-		verificationKey, err := ocpiTokenAuthorization.CreateVerificationKey(*tokenAuthorization)
-
-		if err != nil {
-			metrics.RecordError("OCPI282", "Error creating verification key", err)
-			log.Printf("OCPI282: TokenAuthorization=%#v", tokenAuthorization)
-			return nil, errors.New("error starting session")
-		}
-
 		command, err := r.CommandResolver.StartSession(ctx, credential, *tokenAuthorization, &input.EvseUid)
 
 		if err != nil {
@@ -148,7 +139,7 @@ func (r *RpcCommandResolver) StartSession(reqCtx context.Context, input *ocpirpc
 			return nil, err
 		}
 
-		startResponse := ocpiCommand.NewCommandStartResponse(*command, verificationKey)
+		startResponse := ocpiCommand.NewCommandStartResponse(*command)
 
 		return startResponse, nil
 	}

internal/session/v2.1.1/process.go

@@ -185,7 +185,9 @@ func (r *SessionResolver) ReplaceSessionByIdentifier(ctx context.Context, creden
 		}
 
 		if sessionCreated && session.Status == db.SessionStatusTypePENDING {
-			go r.waitForEvseStatus(credential, session.LocationID, session.EvseID, db.EvseStatusCHARGING, session, session.Status, db.SessionStatusTypeACTIVE, 150)
+			timeout := int(util.GetEnvInt32("WAIT_FOR_EVSE_STATUS_TIMEOUT", 180))
+
+			go r.waitForEvseStatus(credential, session.LocationID, session.EvseID, db.EvseStatusCHARGING, session, session.Status, db.SessionStatusTypeACTIVE, timeout)
 		}
 
 		return &session

internal/session/v2.1.1/sync.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/satimoto/go-datastore/pkg/db"
 	"github.com/satimoto/go-datastore/pkg/util"
+	dto "github.com/satimoto/go-ocpi/internal/dto/v2.1.1"
 	metrics "github.com/satimoto/go-ocpi/internal/metric"
 	coreSession "github.com/satimoto/go-ocpi/internal/session"
 	"github.com/satimoto/go-ocpi/internal/transportation"
@@ -59,43 +60,21 @@ func (r *SessionResolver) SyncByIdentifier(ctx context.Context, credential db.Cr
 	}
 
 	for {
+		if retries >= 5 {
+			break
+		}
+
 		query.Set("limit", fmt.Sprintf("%d", limit))
 		query.Set("offset", fmt.Sprintf("%d", offset))
 		requestUrl.RawQuery = query.Encode()
 
-		response, err := r.OcpiService.Do(http.MethodGet, requestUrl.String(), header, nil)
+		dto, limit := r.sendRequest(requestUrl.String(), header, limit)
 
-		if err != nil {
-			metrics.RecordError("OCPI172", "Error making request", err)
-			log.Printf("OCPI172: Method=%v, Url=%v, Header=%#v", http.MethodGet, requestUrl.String(), header)
+		if dto == nil {
 			retries++
-
-			if retries >= 5 {
-				break
-			}
-
 			continue
 		}
 
-		dto, err := r.UnmarshalPullDto(response.Body)
-		defer response.Body.Close()
-
-		if err != nil {
-			metrics.RecordError("OCPI173", "Error unmarshaling response", err)
-			util.LogHttpResponse("OCPI173", requestUrl.String(), response, true)
-			break
-		}
-
-		limit = transportation.GetXLimitHeader(response, limit)
-
-		if dto.StatusCode != transportation.STATUS_CODE_OK {
-			metrics.RecordError("OCPI174", "Error response failure", err)
-			util.LogHttpRequest("OCPI174", requestUrl.String(), response.Request, true)
-			util.LogHttpResponse("OCPI174", requestUrl.String(), response, true)
-			log.Printf("OCPI174: StatusCode=%v, StatusMessage=%v", dto.StatusCode, dto.StatusMessage)
-			break
-		}
-
 		retries = 0
 
 		if dto.StatusCode == transportation.STATUS_CODE_OK {
@@ -111,3 +90,35 @@ func (r *SessionResolver) SyncByIdentifier(ctx context.Context, credential db.Cr
 		}
 	}
 }
+
+func (r *SessionResolver) sendRequest(url string, header transportation.OcpiRequestHeader, limit int) (*dto.OcpiSessionsDto, int) {
+	response, err := r.OcpiService.Do(http.MethodGet, url, header, nil)
+
+	if err != nil {
+		metrics.RecordError("OCPI172", "Error making request", err)
+		log.Printf("OCPI172: Method=%v, Url=%v, Header=%#v", http.MethodGet, url, header)
+
+		return nil, limit
+	}
+
+	dto, err := r.UnmarshalPullDto(response.Body)
+	defer response.Body.Close()
+
+	if err != nil {
+		metrics.RecordError("OCPI173", "Error unmarshaling response", err)
+		util.LogHttpResponse("OCPI173", url, response, true)
+
+		return nil, limit
+	}
+
+	if dto.StatusCode != transportation.STATUS_CODE_OK {
+		metrics.RecordError("OCPI174", "Error response failure", err)
+		util.LogHttpRequest("OCPI174", url, response.Request, true)
+		util.LogHttpResponse("OCPI174", url, response, true)
+		log.Printf("OCPI174: StatusCode=%v, StatusMessage=%v", dto.StatusCode, dto.StatusMessage)
+
+		return nil, limit
+	}
+
+	return dto, transportation.GetXLimitHeader(response, limit)
+}