You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by sr...@apache.org on 2021/05/01 19:24:10 UTC
[plc4x] branch develop updated: plc4go: implemented write support
for SingleItemRequestInterceptor.go
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/develop by this push:
new fc680ca plc4go: implemented write support for SingleItemRequestInterceptor.go
fc680ca is described below
commit fc680ca2f298f1d58e175ec3cab6f7f3e2c431ed
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Sat May 1 21:23:59 2021 +0200
plc4go: implemented write support for SingleItemRequestInterceptor.go
---
plc4go/internal/plc4go/ads/Connection.go | 2 +
plc4go/internal/plc4go/knxnetip/Connection.go | 2 +
plc4go/internal/plc4go/modbus/Connection.go | 6 +-
.../interceptors/SingleItemRequestInterceptor.go | 85 +++++++++++++++++---
.../plc4go/spi/model/DefaultPlcReadRequest.go | 24 +++---
.../plc4go/spi/model/DefaultPlcWriteRequest.go | 90 ++++++++++++++++++----
.../plc4go/spi/model/DefaultPlcWriteResponse.go | 2 +-
7 files changed, 170 insertions(+), 41 deletions(-)
diff --git a/plc4go/internal/plc4go/ads/Connection.go b/plc4go/internal/plc4go/ads/Connection.go
index c0ddb05..bc290b2 100644
--- a/plc4go/internal/plc4go/ads/Connection.go
+++ b/plc4go/internal/plc4go/ads/Connection.go
@@ -59,7 +59,9 @@ func NewConnection(messageCodec spi.MessageCodec, configuration Configuration, f
messageCodec: messageCodec,
requestInterceptor: interceptors.NewSingleItemRequestInterceptor(
internalModel.NewDefaultPlcReadRequest,
+ internalModel.NewDefaultPlcWriteRequest,
internalModel.NewDefaultPlcReadResponse,
+ internalModel.NewDefaultPlcWriteResponse,
),
reader: &reader,
writer: &writer,
diff --git a/plc4go/internal/plc4go/knxnetip/Connection.go b/plc4go/internal/plc4go/knxnetip/Connection.go
index 14319890..b783884 100644
--- a/plc4go/internal/plc4go/knxnetip/Connection.go
+++ b/plc4go/internal/plc4go/knxnetip/Connection.go
@@ -166,7 +166,9 @@ func NewConnection(transportInstance transports.TransportInstance, options map[s
valueHandler: NewValueHandler(),
requestInterceptor: interceptors.NewSingleItemRequestInterceptor(
internalModel.NewDefaultPlcReadRequest,
+ internalModel.NewDefaultPlcWriteRequest,
internalModel.NewDefaultPlcReadResponse,
+ internalModel.NewDefaultPlcWriteResponse,
),
subscribers: []*Subscriber{},
valueCache: map[uint16][]int8{},
diff --git a/plc4go/internal/plc4go/modbus/Connection.go b/plc4go/internal/plc4go/modbus/Connection.go
index 90aabec..6271ab9 100644
--- a/plc4go/internal/plc4go/modbus/Connection.go
+++ b/plc4go/internal/plc4go/modbus/Connection.go
@@ -48,7 +48,9 @@ func NewConnection(unitIdentifier uint8, messageCodec spi.MessageCodec, options
options: options,
requestInterceptor: interceptors.NewSingleItemRequestInterceptor(
internalModel.NewDefaultPlcReadRequest,
+ internalModel.NewDefaultPlcWriteRequest,
internalModel.NewDefaultPlcReadResponse,
+ internalModel.NewDefaultPlcWriteResponse,
),
}
connection.DefaultConnection = _default.NewDefaultConnection(connection,
@@ -119,11 +121,11 @@ func (m *Connection) ReadRequestBuilder() apiModel.PlcReadRequestBuilder {
}
func (m *Connection) WriteRequestBuilder() apiModel.PlcWriteRequestBuilder {
- // TODO: don't we need a interceptor here?
- return internalModel.NewDefaultPlcWriteRequestBuilder(
+ return internalModel.NewDefaultPlcWriteRequestBuilderWithInterceptor(
m.GetPlcFieldHandler(),
m.GetPlcValueHandler(),
NewWriter(m.unitIdentifier, m.messageCodec),
+ m.requestInterceptor,
)
}
diff --git a/plc4go/internal/plc4go/spi/interceptors/SingleItemRequestInterceptor.go b/plc4go/internal/plc4go/spi/interceptors/SingleItemRequestInterceptor.go
index 442b88d..c3eac4c 100644
--- a/plc4go/internal/plc4go/spi/interceptors/SingleItemRequestInterceptor.go
+++ b/plc4go/internal/plc4go/spi/interceptors/SingleItemRequestInterceptor.go
@@ -32,21 +32,33 @@ type ReaderExposer interface {
GetReader() spi.PlcReader
}
+type WriterExposer interface {
+ GetWriter() spi.PlcWriter
+}
+
type ReadRequestInterceptorExposer interface {
GetReadRequestInterceptor() ReadRequestInterceptor
}
-type requestFactory func(fields map[string]model.PlcField, fieldNames []string, reader spi.PlcReader, readRequestInterceptor ReadRequestInterceptor) model.PlcReadRequest
+type WriteRequestInterceptorExposer interface {
+ GetWriteRequestInterceptor() WriteRequestInterceptor
+}
+
+type readRequestFactory func(fields map[string]model.PlcField, fieldNames []string, reader spi.PlcReader, readRequestInterceptor ReadRequestInterceptor) model.PlcReadRequest
+type writeRequestFactory func(fields map[string]model.PlcField, fieldNames []string, values map[string]values.PlcValue, writer spi.PlcWriter, writeRequestInterceptor WriteRequestInterceptor) model.PlcWriteRequest
-type responseFactory func(request model.PlcReadRequest, responseCodes map[string]model.PlcResponseCode, values map[string]values.PlcValue) model.PlcReadResponse
+type readResponseFactory func(request model.PlcReadRequest, responseCodes map[string]model.PlcResponseCode, values map[string]values.PlcValue) model.PlcReadResponse
+type writeResponseFactory func(request model.PlcWriteRequest, responseCodes map[string]model.PlcResponseCode) model.PlcWriteResponse
type SingleItemRequestInterceptor struct {
- requestFactory requestFactory
- responseFactory responseFactory
+ readRequestFactory readRequestFactory
+ writeRequestFactory writeRequestFactory
+ readResponseFactory readResponseFactory
+ writeResponseFactory writeResponseFactory
}
-func NewSingleItemRequestInterceptor(requestFactory requestFactory, responseFactory responseFactory) SingleItemRequestInterceptor {
- return SingleItemRequestInterceptor{requestFactory, responseFactory}
+func NewSingleItemRequestInterceptor(readRequestFactory readRequestFactory, writeRequestFactory writeRequestFactory, readResponseFactory readResponseFactory, writeResponseFactory writeResponseFactory) SingleItemRequestInterceptor {
+ return SingleItemRequestInterceptor{readRequestFactory, writeRequestFactory, readResponseFactory, writeResponseFactory}
}
func (m SingleItemRequestInterceptor) InterceptReadRequest(readRequest model.PlcReadRequest) []model.PlcReadRequest {
@@ -61,7 +73,7 @@ func (m SingleItemRequestInterceptor) InterceptReadRequest(readRequest model.Plc
for _, fieldName := range readRequest.GetFieldNames() {
log.Debug().Str("fieldName", fieldName).Msg("Splitting into own request")
field := readRequest.GetField(fieldName)
- subReadRequest := m.requestFactory(
+ subReadRequest := m.readRequestFactory(
map[string]model.PlcField{fieldName: field},
[]string{fieldName},
readRequest.(ReaderExposer).GetReader(),
@@ -103,16 +115,65 @@ func (m SingleItemRequestInterceptor) ProcessReadResponses(readRequest model.Plc
}
return model.PlcReadRequestResult{
Request: readRequest,
- Response: m.responseFactory(readRequest, responseCodes, val),
+ Response: m.readResponseFactory(readRequest, responseCodes, val),
Err: err,
}
}
func (m SingleItemRequestInterceptor) InterceptWriteRequest(writeRequest model.PlcWriteRequest) []model.PlcWriteRequest {
- return []model.PlcWriteRequest{writeRequest}
+ // If this request just has one field, go the shortcut
+ if len(writeRequest.GetFieldNames()) == 1 {
+ log.Debug().Msg("We got only one request, no splitting required")
+ return []model.PlcWriteRequest{writeRequest}
+ }
+ log.Trace().Msg("Splitting requests")
+ // In all other cases, create a new write request containing only one item
+ var writeRequests []model.PlcWriteRequest
+ for _, fieldName := range writeRequest.GetFieldNames() {
+ log.Debug().Str("fieldName", fieldName).Msg("Splitting into own request")
+ field := writeRequest.GetField(fieldName)
+ subWriteRequest := m.writeRequestFactory(
+ map[string]model.PlcField{fieldName: field},
+ []string{fieldName},
+ map[string]values.PlcValue{fieldName: writeRequest.GetValue(fieldName)},
+ writeRequest.(WriterExposer).GetWriter(),
+ writeRequest.(WriteRequestInterceptorExposer).GetWriteRequestInterceptor(),
+ )
+ writeRequests = append(writeRequests, subWriteRequest)
+ }
+ return writeRequests
}
-func (m SingleItemRequestInterceptor) ProcessWriteResponses(writeRequest model.PlcWriteRequest, writeResponses []model.PlcWriteRequestResult) model.PlcWriteRequestResult {
- // TODO: unfinished implementation
- return model.PlcWriteRequestResult{}
+func (m SingleItemRequestInterceptor) ProcessWriteResponses(writeRequest model.PlcWriteRequest, writeResults []model.PlcWriteRequestResult) model.PlcWriteRequestResult {
+ if len(writeResults) == 1 {
+ log.Debug().Msg("We got only one response, no merging required")
+ return writeResults[0]
+ }
+ log.Trace().Msg("Merging requests")
+ responseCodes := map[string]model.PlcResponseCode{}
+ var err error = nil
+ for _, writeResult := range writeResults {
+ if writeResult.Err != nil {
+ log.Debug().Err(writeResult.Err).Msgf("Error during write")
+ if err == nil {
+ // Lazy initialization of multi error
+ err = utils.MultiError{MainError: errors.New("while aggregating results"), Errors: []error{writeResult.Err}}
+ } else {
+ multiError := err.(utils.MultiError)
+ multiError.Errors = append(multiError.Errors, writeResult.Err)
+ }
+ } else if writeResult.Response != nil {
+ if len(writeResult.Response.GetRequest().GetFieldNames()) > 1 {
+ log.Error().Int("numberOfFields", len(writeResult.Response.GetRequest().GetFieldNames())).Msg("We should only get 1")
+ }
+ for _, fieldName := range writeResult.Response.GetRequest().GetFieldNames() {
+ responseCodes[fieldName] = writeResult.Response.GetResponseCode(fieldName)
+ }
+ }
+ }
+ return model.PlcWriteRequestResult{
+ Request: writeRequest,
+ Response: m.writeResponseFactory(writeRequest, responseCodes),
+ Err: err,
+ }
}
diff --git a/plc4go/internal/plc4go/spi/model/DefaultPlcReadRequest.go b/plc4go/internal/plc4go/spi/model/DefaultPlcReadRequest.go
index b4cf699..9f2c681 100644
--- a/plc4go/internal/plc4go/spi/model/DefaultPlcReadRequest.go
+++ b/plc4go/internal/plc4go/spi/model/DefaultPlcReadRequest.go
@@ -80,44 +80,44 @@ func (m *DefaultPlcReadRequestBuilder) Build() (model.PlcReadRequest, error) {
type DefaultPlcReadRequest struct {
DefaultRequest
- Reader spi.PlcReader
- ReadRequestInterceptor interceptors.ReadRequestInterceptor
+ reader spi.PlcReader
+ readRequestInterceptor interceptors.ReadRequestInterceptor
}
func NewDefaultPlcReadRequest(fields map[string]model.PlcField, fieldNames []string, reader spi.PlcReader, readRequestInterceptor interceptors.ReadRequestInterceptor) model.PlcReadRequest {
return DefaultPlcReadRequest{
DefaultRequest: NewDefaultRequest(fields, fieldNames),
- Reader: reader,
- ReadRequestInterceptor: readRequestInterceptor,
+ reader: reader,
+ readRequestInterceptor: readRequestInterceptor,
}
}
func (m DefaultPlcReadRequest) GetReader() spi.PlcReader {
- return m.Reader
+ return m.reader
}
func (m DefaultPlcReadRequest) GetReadRequestInterceptor() interceptors.ReadRequestInterceptor {
- return m.ReadRequestInterceptor
+ return m.readRequestInterceptor
}
func (m DefaultPlcReadRequest) Execute() <-chan model.PlcReadRequestResult {
// Shortcut, if no interceptor is defined
- if m.ReadRequestInterceptor == nil {
- return m.Reader.Read(m)
+ if m.readRequestInterceptor == nil {
+ return m.reader.Read(m)
}
// Split the requests up into multiple ones.
- readRequests := m.ReadRequestInterceptor.InterceptReadRequest(m)
+ readRequests := m.readRequestInterceptor.InterceptReadRequest(m)
// Shortcut for single-request-requests
if len(readRequests) == 1 {
- return m.Reader.Read(readRequests[0])
+ return m.reader.Read(readRequests[0])
}
// Create a sub-result-channel slice
var subResultChannels []<-chan model.PlcReadRequestResult
// Iterate over all requests and add the result-channels to the list
for _, subRequest := range readRequests {
- subResultChannels = append(subResultChannels, m.Reader.Read(subRequest))
+ subResultChannels = append(subResultChannels, m.reader.Read(subRequest))
// TODO: Replace this with a real queueing of requests. Later on we need throttling. At the moment this avoids race condition as the read above writes to fast on the line which is a problem for the test
time.Sleep(time.Millisecond * 4)
}
@@ -132,7 +132,7 @@ func (m DefaultPlcReadRequest) Execute() <-chan model.PlcReadRequestResult {
subResults = append(subResults, subResult)
}
// As soon as all are done, process the results
- result := m.ReadRequestInterceptor.ProcessReadResponses(m, subResults)
+ result := m.readRequestInterceptor.ProcessReadResponses(m, subResults)
// Return the final result
resultChannel <- result
}()
diff --git a/plc4go/internal/plc4go/spi/model/DefaultPlcWriteRequest.go b/plc4go/internal/plc4go/spi/model/DefaultPlcWriteRequest.go
index ae5eb75..e9e6a6e 100644
--- a/plc4go/internal/plc4go/spi/model/DefaultPlcWriteRequest.go
+++ b/plc4go/internal/plc4go/spi/model/DefaultPlcWriteRequest.go
@@ -22,21 +22,24 @@ package model
import (
"encoding/xml"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi"
+ "github.com/apache/plc4x/plc4go/internal/plc4go/spi/interceptors"
values2 "github.com/apache/plc4x/plc4go/internal/plc4go/spi/values"
"github.com/apache/plc4x/plc4go/pkg/plc4go/model"
"github.com/apache/plc4x/plc4go/pkg/plc4go/values"
"github.com/pkg/errors"
+ "time"
)
type DefaultPlcWriteRequestBuilder struct {
- writer spi.PlcWriter
- fieldHandler spi.PlcFieldHandler
- valueHandler spi.PlcValueHandler
- queries map[string]string
- queryNames []string
- fields map[string]model.PlcField
- fieldNames []string
- values map[string]interface{}
+ writer spi.PlcWriter
+ fieldHandler spi.PlcFieldHandler
+ valueHandler spi.PlcValueHandler
+ queries map[string]string
+ queryNames []string
+ fields map[string]model.PlcField
+ fieldNames []string
+ values map[string]interface{}
+ writeRequestInterceptor interceptors.WriteRequestInterceptor
}
func NewDefaultPlcWriteRequestBuilder(fieldHandler spi.PlcFieldHandler, valueHandler spi.PlcValueHandler, writer spi.PlcWriter) *DefaultPlcWriteRequestBuilder {
@@ -52,6 +55,28 @@ func NewDefaultPlcWriteRequestBuilder(fieldHandler spi.PlcFieldHandler, valueHan
}
}
+func NewDefaultPlcWriteRequestBuilderWithInterceptor(fieldHandler spi.PlcFieldHandler, valueHandler spi.PlcValueHandler, writer spi.PlcWriter, writeRequestInterceptor interceptors.WriteRequestInterceptor) *DefaultPlcWriteRequestBuilder {
+ return &DefaultPlcWriteRequestBuilder{
+ writer: writer,
+ fieldHandler: fieldHandler,
+ valueHandler: valueHandler,
+ queries: map[string]string{},
+ queryNames: make([]string, 0),
+ fields: map[string]model.PlcField{},
+ fieldNames: make([]string, 0),
+ values: map[string]interface{}{},
+ writeRequestInterceptor: writeRequestInterceptor,
+ }
+}
+
+func (m *DefaultPlcWriteRequestBuilder) GetWriter() spi.PlcWriter {
+ return m.writer
+}
+
+func (m *DefaultPlcWriteRequestBuilder) GetWriteRequestInterceptor() interceptors.WriteRequestInterceptor {
+ return m.writeRequestInterceptor
+}
+
func (m *DefaultPlcWriteRequestBuilder) AddQuery(name string, query string, value interface{}) model.PlcWriteRequestBuilder {
m.queryNames = append(m.queryNames, name)
m.queries[name] = query
@@ -86,21 +111,58 @@ func (m *DefaultPlcWriteRequestBuilder) Build() (model.PlcWriteRequest, error) {
}
plcValues[name] = value
}
- return NewDefaultPlcWriteRequest(m.fields, m.fieldNames, plcValues, m.writer), nil
+ return NewDefaultPlcWriteRequest(m.fields, m.fieldNames, plcValues, m.writer, m.writeRequestInterceptor), nil
}
type DefaultPlcWriteRequest struct {
DefaultRequest
- values map[string]values.PlcValue
- writer spi.PlcWriter
+ values map[string]values.PlcValue
+ writer spi.PlcWriter
+ writeRequestInterceptor interceptors.WriteRequestInterceptor
}
-func NewDefaultPlcWriteRequest(fields map[string]model.PlcField, fieldNames []string, values map[string]values.PlcValue, writer spi.PlcWriter) model.PlcWriteRequest {
- return DefaultPlcWriteRequest{NewDefaultRequest(fields, fieldNames), values, writer}
+func NewDefaultPlcWriteRequest(fields map[string]model.PlcField, fieldNames []string, values map[string]values.PlcValue, writer spi.PlcWriter, writeRequestInterceptor interceptors.WriteRequestInterceptor) model.PlcWriteRequest {
+ return DefaultPlcWriteRequest{NewDefaultRequest(fields, fieldNames), values, writer, writeRequestInterceptor}
}
func (m DefaultPlcWriteRequest) Execute() <-chan model.PlcWriteRequestResult {
- return m.writer.Write(m)
+ // Shortcut, if no interceptor is defined
+ if m.writeRequestInterceptor == nil {
+ return m.writer.Write(m)
+ }
+
+ // Split the requests up into multiple ones.
+ writeRequests := m.writeRequestInterceptor.InterceptWriteRequest(m)
+ // Shortcut for single-request-requests
+ if len(writeRequests) == 1 {
+ return m.writer.Write(writeRequests[0])
+ }
+ // Create a sub-result-channel slice
+ var subResultChannels []<-chan model.PlcWriteRequestResult
+
+ // Iterate over all requests and add the result-channels to the list
+ for _, subRequest := range writeRequests {
+ subResultChannels = append(subResultChannels, m.writer.Write(subRequest))
+ // TODO: Replace this with a real queueing of requests. Later on we need throttling. At the moment this avoids race condition as the read above writes to fast on the line which is a problem for the test
+ time.Sleep(time.Millisecond * 4)
+ }
+
+ // Create a new result-channel, which completes as soon as all sub-result-channels have returned
+ resultChannel := make(chan model.PlcWriteRequestResult)
+ go func() {
+ var subResults []model.PlcWriteRequestResult
+ // Iterate over all sub-results
+ for _, subResultChannel := range subResultChannels {
+ subResult := <-subResultChannel
+ subResults = append(subResults, subResult)
+ }
+ // As soon as all are done, process the results
+ result := m.writeRequestInterceptor.ProcessWriteResponses(m, subResults)
+ // Return the final result
+ resultChannel <- result
+ }()
+
+ return resultChannel
}
func (m DefaultPlcWriteRequest) GetValue(name string) values.PlcValue {
diff --git a/plc4go/internal/plc4go/spi/model/DefaultPlcWriteResponse.go b/plc4go/internal/plc4go/spi/model/DefaultPlcWriteResponse.go
index c5fe0c3..83584c0 100644
--- a/plc4go/internal/plc4go/spi/model/DefaultPlcWriteResponse.go
+++ b/plc4go/internal/plc4go/spi/model/DefaultPlcWriteResponse.go
@@ -29,7 +29,7 @@ type DefaultPlcWriteResponse struct {
request model.PlcWriteRequest
}
-func NewDefaultPlcWriteResponse(request model.PlcWriteRequest, responseCodes map[string]model.PlcResponseCode) DefaultPlcWriteResponse {
+func NewDefaultPlcWriteResponse(request model.PlcWriteRequest, responseCodes map[string]model.PlcResponseCode) model.PlcWriteResponse {
return DefaultPlcWriteResponse{
DefaultResponse: NewDefaultResponse(responseCodes),
request: request,