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/04/07 14:33:28 UTC
[plc4x] branch develop updated (9891c3d -> 6246432)
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a change to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git.
from 9891c3d - Fixed the Log-Level setting
new e254ea6 plc4go: fix log levels
new 6246432 plc4go: refactored message codes into DefaultMessageCodec
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../main/drivers/tests/manual_ads_driver_test.go | 17 +-
.../hello_world_plc4go_knx_read_group_address.go | 4 +
.../hello_world_plc4go_knx_subscription.go | 4 +
plc4go/internal/plc4go/ads/MessageCodec.go | 165 +--------------
plc4go/internal/plc4go/knxnetip/Connection.go | 4 +
plc4go/internal/plc4go/knxnetip/Driver.go | 3 +-
plc4go/internal/plc4go/knxnetip/MessageCodec.go | 226 ++++-----------------
plc4go/internal/plc4go/modbus/MessageCodec.go | 167 +--------------
plc4go/internal/plc4go/spi/MessageCodec.go | 220 +++++++++++++++++++-
plc4go/pkg/plc4go/logging/init.go | 7 +-
10 files changed, 308 insertions(+), 509 deletions(-)
[plc4x] 01/02: plc4go: fix log levels
Posted by sr...@apache.org.
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
commit e254ea6b8146c7b8aec85dfba86282f2d110ca1d
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Wed Apr 7 14:59:33 2021 +0200
plc4go: fix log levels
---
plc4go/pkg/plc4go/logging/init.go | 7 ++++++-
1 file changed, 6 insertions(+), 1 deletion(-)
diff --git a/plc4go/pkg/plc4go/logging/init.go b/plc4go/pkg/plc4go/logging/init.go
index b41f90b..a02f14e 100644
--- a/plc4go/pkg/plc4go/logging/init.go
+++ b/plc4go/pkg/plc4go/logging/init.go
@@ -46,11 +46,16 @@ func InfoLevel() {
log.Logger = log.Logger.Level(zerolog.InfoLevel)
}
-// DebugLevel configures zerolog to InfoLevel
+// DebugLevel configures zerolog to DebugLevel
func DebugLevel() {
log.Logger = log.Logger.Level(zerolog.DebugLevel)
}
+// TraceLevel configures zerolog to TraceLevel
+func TraceLevel() {
+ log.Logger = log.Logger.Level(zerolog.TraceLevel)
+}
+
// ResetLogging can be used to reset to the old log settings
func ResetLogging() {
log.Logger = oldLogger
[plc4x] 02/02: plc4go: refactored message codes into
DefaultMessageCodec
Posted by sr...@apache.org.
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
commit 6246432e5ac082631b06c61bc1d733e7fab406ef
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Wed Apr 7 16:33:01 2021 +0200
plc4go: refactored message codes into DefaultMessageCodec
---
.../main/drivers/tests/manual_ads_driver_test.go | 17 +-
.../hello_world_plc4go_knx_read_group_address.go | 4 +
.../hello_world_plc4go_knx_subscription.go | 4 +
plc4go/internal/plc4go/ads/MessageCodec.go | 165 +--------------
plc4go/internal/plc4go/knxnetip/Connection.go | 4 +
plc4go/internal/plc4go/knxnetip/Driver.go | 3 +-
plc4go/internal/plc4go/knxnetip/MessageCodec.go | 226 ++++-----------------
plc4go/internal/plc4go/modbus/MessageCodec.go | 167 +--------------
plc4go/internal/plc4go/spi/MessageCodec.go | 220 +++++++++++++++++++-
9 files changed, 302 insertions(+), 508 deletions(-)
diff --git a/plc4go/cmd/main/drivers/tests/manual_ads_driver_test.go b/plc4go/cmd/main/drivers/tests/manual_ads_driver_test.go
index 14d5173..5012307 100644
--- a/plc4go/cmd/main/drivers/tests/manual_ads_driver_test.go
+++ b/plc4go/cmd/main/drivers/tests/manual_ads_driver_test.go
@@ -20,6 +20,8 @@
package tests
import (
+ "fmt"
+ _ "github.com/apache/plc4x/plc4go/cmd/main/initializetest"
"github.com/apache/plc4x/plc4go/internal/plc4go/ads"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi/testutils"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi/transports/tcp"
@@ -28,7 +30,7 @@ import (
)
func TestManualAds(t *testing.T) {
- t.Skip()
+ //t.Skip()
/*
* Test program code on the PLC with the test-data.
@@ -69,10 +71,21 @@ func TestManualAds(t *testing.T) {
*
*/
+ spsIp := "192.168.23.20"
+ /////
+ // TODO: adjust this to your ip address
+ clientIp := "192.168.24.1"
+ //
+ ////
+ sourceAmsNetId := clientIp + ".1.1"
+ sourceAmsPort := 65534
+ targetAmsNetId := spsIp + ".1.1"
+ targetAmsPort := 851
+ connectionString := fmt.Sprintf("ads:tcp://%s?sourceAmsNetId=%s&sourceAmsPort=%d&targetAmsNetId=%s&targetAmsPort=%d", spsIp, sourceAmsNetId, sourceAmsPort, targetAmsNetId, targetAmsPort)
driverManager := plc4go.NewPlcDriverManager()
driverManager.RegisterDriver(ads.NewDriver())
driverManager.RegisterTransport(tcp.NewTransport())
- test := testutils.NewManualTestSuite("ads:tcp://192.168.23.20?sourceAmsNetId=192.168.23.200.1.1&sourceAmsPort=65534&targetAmsNetId=192.168.23.20.1.1&targetAmsPort=851", driverManager)
+ test := testutils.NewManualTestSuite(connectionString, driverManager)
test.AddTestCase("main.hurz_BOOL:BOOL", true)
test.AddTestCase("main.hurz_BYTE:BYTE", []bool{false, false, true, false, true, false, true, false})
test.AddTestCase("main.hurz_WORD:WORD", []bool{true, false, true, false, false, true, false, true, true, false, true, true, true, false, false, false})
diff --git a/plc4go/examples/knx/groupAddressRead/hello_world_plc4go_knx_read_group_address.go b/plc4go/examples/knx/groupAddressRead/hello_world_plc4go_knx_read_group_address.go
index 46c2ff3..418917a 100644
--- a/plc4go/examples/knx/groupAddressRead/hello_world_plc4go_knx_read_group_address.go
+++ b/plc4go/examples/knx/groupAddressRead/hello_world_plc4go_knx_read_group_address.go
@@ -22,10 +22,14 @@ import (
"fmt"
"github.com/apache/plc4x/plc4go/pkg/plc4go"
"github.com/apache/plc4x/plc4go/pkg/plc4go/drivers"
+ "github.com/apache/plc4x/plc4go/pkg/plc4go/logging"
"github.com/apache/plc4x/plc4go/pkg/plc4go/model"
)
func main() {
+ // Set logging to INFO
+ logging.InfoLevel()
+
driverManager := plc4go.NewPlcDriverManager()
drivers.RegisterKnxDriver(driverManager)
diff --git a/plc4go/examples/knx/subscribe/hello_world_plc4go_knx_subscription.go b/plc4go/examples/knx/subscribe/hello_world_plc4go_knx_subscription.go
index c8821d4..5dfcba1 100644
--- a/plc4go/examples/knx/subscribe/hello_world_plc4go_knx_subscription.go
+++ b/plc4go/examples/knx/subscribe/hello_world_plc4go_knx_subscription.go
@@ -22,6 +22,7 @@ import (
"fmt"
"github.com/apache/plc4x/plc4go/pkg/plc4go"
"github.com/apache/plc4x/plc4go/pkg/plc4go/drivers"
+ "github.com/apache/plc4x/plc4go/pkg/plc4go/logging"
"github.com/apache/plc4x/plc4go/pkg/plc4go/model"
"github.com/apache/plc4x/plc4go/pkg/plc4go/values"
"strings"
@@ -29,6 +30,9 @@ import (
)
func main() {
+ // Set logging to INFO
+ logging.InfoLevel()
+
driverManager := plc4go.NewPlcDriverManager()
drivers.RegisterKnxDriver(driverManager)
diff --git a/plc4go/internal/plc4go/ads/MessageCodec.go b/plc4go/internal/plc4go/ads/MessageCodec.go
index 952fa44..cb6c716 100644
--- a/plc4go/internal/plc4go/ads/MessageCodec.go
+++ b/plc4go/internal/plc4go/ads/MessageCodec.go
@@ -19,68 +19,26 @@
package ads
import (
- "fmt"
"github.com/apache/plc4x/plc4go/internal/plc4go/ads/readwrite/model"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi"
- "github.com/apache/plc4x/plc4go/internal/plc4go/spi/plcerrors"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi/transports"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi/utils"
"github.com/pkg/errors"
"github.com/rs/zerolog/log"
- "time"
)
-type Expectation struct {
- expiration time.Time
- acceptsMessage spi.AcceptsMessage
- handleMessage spi.HandleMessage
- handleError spi.HandleError
-}
-
-func (m Expectation) String() string {
- return fmt.Sprintf("Expectation(expires at %v)", m.expiration)
-}
-
type MessageCodec struct {
- expectationCounter int32
- transportInstance transports.TransportInstance
- defaultIncomingMessageChannel chan interface{}
- expectations []Expectation
- running bool
+ *spi.DefaultCodec
}
func NewMessageCodec(transportInstance transports.TransportInstance) *MessageCodec {
codec := &MessageCodec{
- expectationCounter: 1,
- transportInstance: transportInstance,
- defaultIncomingMessageChannel: make(chan interface{}),
- expectations: []Expectation{},
- running: true,
+ DefaultCodec: spi.NewDefaultCodec(transportInstance),
}
- // TODO: should we better move this go func into Connect(). If not a better explanation why we start the worker so early
- // Start a worker that handles processing of responses
- go work(codec)
+ codec.DefaultCodecRequiredInterface = codec
return codec
}
-func (m *MessageCodec) Connect() error {
- log.Info().Msg("Connecting")
- err := m.transportInstance.Connect()
- if err == nil {
- if !m.running {
- go work(m)
- }
- m.running = true
- }
- return err
-}
-
-func (m *MessageCodec) Disconnect() error {
- log.Info().Msg("Disconnecting")
- m.running = false
- return m.transportInstance.Close()
-}
-
func (m *MessageCodec) Send(message interface{}) error {
log.Trace().Msg("Sending message")
// Cast the message to the correct type of struct
@@ -93,44 +51,19 @@ func (m *MessageCodec) Send(message interface{}) error {
}
// Send it to the PLC
- err = m.transportInstance.Write(wb.GetBytes())
+ err = m.TransportInstance.Write(wb.GetBytes())
if err != nil {
return errors.Wrap(err, "error sending request")
}
return nil
}
-func (m *MessageCodec) Expect(acceptsMessage spi.AcceptsMessage, handleMessage spi.HandleMessage, handleError spi.HandleError, ttl time.Duration) error {
- expectation := Expectation{
- expiration: time.Now().Add(ttl),
- acceptsMessage: acceptsMessage,
- handleMessage: handleMessage,
- handleError: handleError,
- }
- m.expectations = append(m.expectations, expectation)
- return nil
-}
-
-func (m *MessageCodec) SendRequest(message interface{}, acceptsMessage spi.AcceptsMessage, handleMessage spi.HandleMessage, handleError spi.HandleError, ttl time.Duration) error {
- log.Trace().Msg("Sending request")
- // Send the actual message
- err := m.Send(message)
- if err != nil {
- return errors.Wrap(err, "Error sending the request")
- }
- return m.Expect(acceptsMessage, handleMessage, handleError, ttl)
-}
-
-func (m *MessageCodec) GetDefaultIncomingMessageChannel() chan interface{} {
- return m.defaultIncomingMessageChannel
-}
-
-func (m *MessageCodec) receive() (interface{}, error) {
+func (m *MessageCodec) Receive() (interface{}, error) {
log.Trace().Msg("receiving")
// We need at least 6 bytes in order to know how big the packet is in total
- if num, err := m.transportInstance.GetNumReadableBytes(); (err == nil) && (num >= 6) {
+ if num, err := m.TransportInstance.GetNumReadableBytes(); (err == nil) && (num >= 6) {
log.Debug().Msgf("we got %d readable bytes", num)
- data, err := m.transportInstance.PeekReadableBytes(6)
+ data, err := m.TransportInstance.PeekReadableBytes(6)
if err != nil {
log.Warn().Err(err).Msg("error peeking")
// TODO: Possibly clean up ...
@@ -143,7 +76,7 @@ func (m *MessageCodec) receive() (interface{}, error) {
log.Debug().Msgf("Not enough bytes. Got: %d Need: %d\n", num, packetSize)
return nil, nil
}
- data, err = m.transportInstance.Read(packetSize)
+ data, err = m.TransportInstance.Read(packetSize)
if err != nil {
// TODO: Possibly clean up ...
return nil, nil
@@ -163,85 +96,3 @@ func (m *MessageCodec) receive() (interface{}, error) {
// TODO: maybe we return here a not enough error error
return nil, nil
}
-
-func work(m *MessageCodec) {
- // Start an endless loop
-mainLoop:
- for m.running {
- // Check for any expired expectations.
- // (Doing this outside the loop lets us expire expectations even if no input is coming in)
- now := time.Now()
-
- // Guard against empty expectations
- if len(m.expectations) <= 0 {
- log.Trace().Msg("we got no expectation")
- // Sleep for 10ms
- time.Sleep(time.Millisecond * 10)
- continue mainLoop
- }
- expectationLoop:
- for index, expectation := range m.expectations {
- // Check if this expectation has expired.
- if now.After(expectation.expiration) {
- // Remove this expectation from the list.
- m.expectations = append(m.expectations[:index], m.expectations[index+1:]...)
- // Call the error handler.
- err := expectation.handleError(plcerrors.NewTimeoutError(now.Sub(expectation.expiration)))
- if err != nil {
- log.Error().Err(err).Msg("Got an error handling error on expectation")
- }
- continue expectationLoop
- }
- }
-
- // Check for incoming messages.
- message, err := m.receive()
- if err != nil {
- log.Error().Err(err).Msg("got an error reading from transport")
- time.Sleep(time.Millisecond * 10)
- continue mainLoop
- }
- if message == nil {
- // Sleep for 10ms before checking again, in order to not
- // consume 100% CPU Power.
- time.Sleep(time.Millisecond * 10)
- continue mainLoop
- }
-
- // Go through all expectations
- messageHandled := false
- for index, expectation := range m.expectations {
- // Check if the current message matches the expectations
- // If it does, let it handle the message.
- if accepts := expectation.acceptsMessage(message); accepts {
- log.Debug().Stringer("expectation", expectation).Msg("accepts message")
- err = expectation.handleMessage(message)
- if err == nil {
- messageHandled = true
- // If this is the last element of the list remove it differently than if it's before that
- if (index + 1) == len(m.expectations) {
- m.expectations = m.expectations[:index]
- } else if (index + 1) < len(m.expectations) {
- m.expectations = append(m.expectations[:index], m.expectations[index+1:]...)
- }
- } else {
- // Pass the error to the error handler.
- err := expectation.handleError(err)
- if err != nil {
- log.Error().Err(err).Msg("Got an error handling error on expectation")
- }
- }
- }
- }
-
- // If the message has not been handled and a default handler is provided, call this ...
- if !messageHandled {
- // TODO: how do we prevent endless blocking if there is no reader on this channel?
- m.defaultIncomingMessageChannel <- message
- }
- }
-}
-
-func (m MessageCodec) GetTransportInstance() transports.TransportInstance {
- return m.transportInstance
-}
diff --git a/plc4go/internal/plc4go/knxnetip/Connection.go b/plc4go/internal/plc4go/knxnetip/Connection.go
index 4d45325..a76dd06 100644
--- a/plc4go/internal/plc4go/knxnetip/Connection.go
+++ b/plc4go/internal/plc4go/knxnetip/Connection.go
@@ -134,6 +134,10 @@ type Connection struct {
sync.Mutex
}
+func (m *Connection) String() string {
+ return fmt.Sprintf("knx.Connection{}")
+}
+
type KnxReadResult struct {
value *values.PlcValue
numItems uint8
diff --git a/plc4go/internal/plc4go/knxnetip/Driver.go b/plc4go/internal/plc4go/knxnetip/Driver.go
index 8bef0c9..10ff8c9 100644
--- a/plc4go/internal/plc4go/knxnetip/Driver.go
+++ b/plc4go/internal/plc4go/knxnetip/Driver.go
@@ -24,6 +24,7 @@ import (
"github.com/apache/plc4x/plc4go/pkg/plc4go"
"github.com/apache/plc4x/plc4go/pkg/plc4go/model"
"github.com/pkg/errors"
+ "github.com/rs/zerolog/log"
"net/url"
)
@@ -78,7 +79,7 @@ func (m Driver) GetConnection(transportUrl url.URL, transports map[string]transp
// Create the new connection
connection := NewConnection(transportInstance, options, m.fieldHandler)
-
+ log.Info().Stringer("connection", connection).Msg("created connection, connecting now")
return connection.Connect()
}
diff --git a/plc4go/internal/plc4go/knxnetip/MessageCodec.go b/plc4go/internal/plc4go/knxnetip/MessageCodec.go
index a711ad1..4fb531e 100644
--- a/plc4go/internal/plc4go/knxnetip/MessageCodec.go
+++ b/plc4go/internal/plc4go/knxnetip/MessageCodec.go
@@ -19,78 +19,30 @@
package knxnetip
import (
- "fmt"
"github.com/apache/plc4x/plc4go/internal/plc4go/knxnetip/readwrite/model"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi"
- "github.com/apache/plc4x/plc4go/internal/plc4go/spi/plcerrors"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi/transports"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi/utils"
"github.com/pkg/errors"
"github.com/rs/zerolog/log"
- "time"
)
-type Expectation struct {
- expiration time.Time
- acceptsMessage spi.AcceptsMessage
- handleMessage spi.HandleMessage
- handleError spi.HandleError
-}
-
-func (m Expectation) String() string {
- return fmt.Sprintf("Expectation(expires at %v)", m.expiration)
-}
-
type MessageCodec struct {
- sequenceCounter int32
- transportInstance transports.TransportInstance
- messageInterceptor func(message interface{})
- defaultIncomingMessageChannel chan interface{}
- expectations []Expectation
- running bool
+ *spi.DefaultCodec
+ sequenceCounter int32
+ messageInterceptor func(message interface{})
}
func NewMessageCodec(transportInstance transports.TransportInstance, messageInterceptor func(message interface{})) *MessageCodec {
codec := &MessageCodec{
- sequenceCounter: 0,
- transportInstance: transportInstance,
- messageInterceptor: messageInterceptor,
- defaultIncomingMessageChannel: make(chan interface{}),
- expectations: []Expectation{},
- running: true,
+ DefaultCodec: spi.NewDefaultCodec(transportInstance),
+ messageInterceptor: messageInterceptor,
}
- // TODO: should we better move this go func into Connect(). If not a better explanation why we start the worker so early
- // Start a worker that handles processing of responses
- go func() {
- for {
- work(codec)
- // TODO: smells like antipattern: why would the loop exit randomly? because of panic? better catch them in this case
- log.Warn().Msg("Restarting Work loop")
- }
- }()
+ codec.DefaultCodecRequiredInterface = codec
+ codec.CustomMessageHandling = CustomMessageHandling
return codec
}
-func (m *MessageCodec) Connect() error {
- log.Info().Msg("Connecting")
- // "connect" to the remote UDP server
- err := m.transportInstance.Connect()
- if err == nil {
- // TODO: smells like antipattern: why would the loop exit randomly? because of panic? better catch them in this case
- /*if !m.running {
- go work(m)
- }*/
- m.running = true
- }
- return err
-}
-
-func (m *MessageCodec) Disconnect() error {
- log.Info().Msg("Disconnecting")
- m.running = false
- return m.transportInstance.Close()
-}
-
func (m *MessageCodec) Send(message interface{}) error {
log.Trace().Msg("Sending message")
// Cast the message to the correct type of struct
@@ -103,44 +55,19 @@ func (m *MessageCodec) Send(message interface{}) error {
}
// Send it to the PLC
- err = m.transportInstance.Write(wb.GetBytes())
+ err = m.TransportInstance.Write(wb.GetBytes())
if err != nil {
return errors.Wrap(err, "error sending request ")
}
return nil
}
-func (m *MessageCodec) Expect(acceptsMessage spi.AcceptsMessage, handleMessage spi.HandleMessage, handleError spi.HandleError, ttl time.Duration) error {
- expectation := Expectation{
- expiration: time.Now().Add(ttl),
- acceptsMessage: acceptsMessage,
- handleMessage: handleMessage,
- handleError: handleError,
- }
- m.expectations = append(m.expectations, expectation)
- return nil
-}
-
-func (m *MessageCodec) SendRequest(message interface{}, acceptsMessage spi.AcceptsMessage, handleMessage spi.HandleMessage, handleError spi.HandleError, ttl time.Duration) error {
- log.Trace().Msg("Sending request")
- // Send the actual message
- err := m.Send(message)
- if err != nil {
- return errors.Wrap(err, "Error sending the request")
- }
- return m.Expect(acceptsMessage, handleMessage, handleError, ttl)
-}
-
-func (m *MessageCodec) GetDefaultIncomingMessageChannel() chan interface{} {
- return m.defaultIncomingMessageChannel
-}
-
-func (m *MessageCodec) receive() (interface{}, error) {
+func (m *MessageCodec) Receive() (interface{}, error) {
log.Trace().Msg("receiving")
// We need at least 6 bytes in order to know how big the packet is in total
- if num, err := m.transportInstance.GetNumReadableBytes(); (err == nil) && (num >= 6) {
+ if num, err := m.TransportInstance.GetNumReadableBytes(); (err == nil) && (num >= 6) {
log.Debug().Msgf("we got %d readable bytes", num)
- data, err := m.transportInstance.PeekReadableBytes(6)
+ data, err := m.TransportInstance.PeekReadableBytes(6)
if err != nil {
log.Warn().Err(err).Msg("error peeking")
// TODO: Possibly clean up ...
@@ -152,7 +79,7 @@ func (m *MessageCodec) receive() (interface{}, error) {
log.Debug().Msgf("Not enough bytes. Got: %d Need: %d\n", num, packetSize)
return nil, nil
}
- data, err = m.transportInstance.Read(packetSize)
+ data, err = m.TransportInstance.Read(packetSize)
if err != nil {
log.Warn().Err(err).Msg("error reading")
// TODO: Possibly clean up ...
@@ -173,114 +100,33 @@ func (m *MessageCodec) receive() (interface{}, error) {
return nil, nil
}
-func work(m *MessageCodec) {
- // If this method ever exits, start it again.
- // TODO: smells like antipattern: why would the loop exit randomly? because of panic? better catch them in this case
- defer work(m)
- // Start an endless loop
-mainLoop:
- for m.running {
- // Check for any expired expectations.
- // (Doing this outside the loop lets us expire expectations even if no input is coming in)
- now := time.Now()
-
- // Guard against empty expectations
- if len(m.expectations) <= 0 {
- log.Trace().Msg("we got no expectation")
- // Sleep for 10ms
- time.Sleep(time.Millisecond * 10)
- continue mainLoop
- }
- expectationLoop:
- for index, expectation := range m.expectations {
- // Check if this expectation has expired.
- if now.After(expectation.expiration) {
- // Remove this expectation from the list.
- m.expectations = append(m.expectations[:index], m.expectations[index+1:]...)
- // Call the error handler.
- err := expectation.handleError(plcerrors.NewTimeoutError(now.Sub(expectation.expiration)))
- if err != nil {
- log.Error().Err(err).Msg("Got an error handling error on expectation")
- }
- continue expectationLoop
- }
- }
+func CustomMessageHandling(codec *spi.DefaultCodecRequiredInterface, message interface{}) bool {
+ // If this message is a simple KNXNet/IP UDP Ack, ignore it for now
+ tunnelingResponse := model.CastTunnelingResponse(message)
+ if tunnelingResponse != nil {
+ return true
+ }
- // Check for incoming messages.
- message, err := m.receive()
+ // If this is an incoming tunneling request, automatically send a tunneling ACK back to the gateway
+ tunnelingRequest := model.CastTunnelingRequest(message)
+ if tunnelingRequest != nil {
+ response := model.NewTunnelingResponse(
+ model.NewTunnelingResponseDataBlock(
+ tunnelingRequest.TunnelingRequestDataBlock.CommunicationChannelId,
+ tunnelingRequest.TunnelingRequestDataBlock.SequenceCounter,
+ model.Status_NO_ERROR),
+ )
+ err := (*codec).Send(response)
if err != nil {
- log.Error().Err(err).Msg("got an error reading from transport")
- time.Sleep(time.Millisecond * 10)
- continue mainLoop
- }
- if message == nil {
- // Sleep for 10ms before checking again, in order to not
- // consume 100% CPU Power.
- time.Sleep(time.Millisecond * 10)
- continue mainLoop
- }
-
- // If this message is a simple KNXNet/IP UDP Ack, ignore it for now
- tunnelingResponse := model.CastTunnelingResponse(message)
- if tunnelingResponse != nil {
- continue mainLoop
- }
-
- // If this is an incoming tunneling request, automatically send a tunneling ACK back to the gateway
- tunnelingRequest := model.CastTunnelingRequest(message)
- if tunnelingRequest != nil {
- response := model.NewTunnelingResponse(
- model.NewTunnelingResponseDataBlock(
- tunnelingRequest.TunnelingRequestDataBlock.CommunicationChannelId,
- tunnelingRequest.TunnelingRequestDataBlock.SequenceCounter,
- model.Status_NO_ERROR),
- )
- err = m.Send(response)
- if err != nil {
- log.Warn().Err(err).Msg("got an error sending ACK from transport")
- }
- }
-
- // Handle the packet itself
- // Give a message interceptor a chance to intercept
- if m.messageInterceptor != nil {
- m.messageInterceptor(message)
- }
-
- // Go through all expectations
- messageHandled := false
- for index, expectation := range m.expectations {
- // Check if the current message matches the expectations
- // If it does, let it handle the message.
- if accepts := expectation.acceptsMessage(message); accepts {
- log.Debug().Stringer("expectation", expectation).Msg("accepts message")
- err = expectation.handleMessage(message)
- if err == nil {
- messageHandled = true
- // If this is the last element of the list remove it differently than if it's before that
- if (index + 1) == len(m.expectations) {
- m.expectations = m.expectations[:index]
- } else if (index + 1) < len(m.expectations) {
- m.expectations = append(m.expectations[:index], m.expectations[index+1:]...)
- }
- } else {
- // Pass the error to the error handler.
- err := expectation.handleError(err)
- if err != nil {
- log.Error().Err(err).Msg("Got an error handling error on expectation")
- }
- }
- }
- }
-
- // If the message has not been handled and a default handler is provided, call this ...
- if !messageHandled {
- // TODO: how do we prevent endless blocking if there is no reader on this channel?
- m.defaultIncomingMessageChannel <- message
+ log.Warn().Err(err).Msg("got an error sending ACK from transport")
}
}
-}
-func (m MessageCodec) GetTransportInstance() transports.TransportInstance {
- return m.transportInstance
+ localCodec := (*codec).(*MessageCodec)
+ // Handle the packet itself
+ // Give a message interceptor a chance to intercept
+ if (*localCodec).messageInterceptor != nil {
+ (*localCodec).messageInterceptor(message)
+ }
+ return false
}
diff --git a/plc4go/internal/plc4go/modbus/MessageCodec.go b/plc4go/internal/plc4go/modbus/MessageCodec.go
index a11027b..ea55712 100644
--- a/plc4go/internal/plc4go/modbus/MessageCodec.go
+++ b/plc4go/internal/plc4go/modbus/MessageCodec.go
@@ -19,68 +19,28 @@
package modbus
import (
- "fmt"
"github.com/apache/plc4x/plc4go/internal/plc4go/modbus/readwrite/model"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi"
- "github.com/apache/plc4x/plc4go/internal/plc4go/spi/plcerrors"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi/transports"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi/utils"
"github.com/pkg/errors"
"github.com/rs/zerolog/log"
- "time"
)
-type Expectation struct {
- expiration time.Time
- acceptsMessage spi.AcceptsMessage
- handleMessage spi.HandleMessage
- handleError spi.HandleError
-}
-
-func (m Expectation) String() string {
- return fmt.Sprintf("Expectation(expires at %v)", m.expiration)
-}
-
type MessageCodec struct {
- expectationCounter int32
- transportInstance transports.TransportInstance
- defaultIncomingMessageChannel chan interface{}
- expectations []Expectation
- running bool
+ *spi.DefaultCodec
+ expectationCounter int32
}
func NewMessageCodec(transportInstance transports.TransportInstance) *MessageCodec {
codec := &MessageCodec{
- expectationCounter: 1,
- transportInstance: transportInstance,
- defaultIncomingMessageChannel: make(chan interface{}),
- expectations: []Expectation{},
- running: true,
+ DefaultCodec: spi.NewDefaultCodec(transportInstance),
+ expectationCounter: 1,
}
- // TODO: should we better move this go func into Connect(). If not a better explanation why we start the worker so early
- // Start a worker that handles processing of responses
- go work(codec)
+ codec.DefaultCodecRequiredInterface = codec
return codec
}
-func (m *MessageCodec) Connect() error {
- log.Info().Msg("Connecting")
- err := m.transportInstance.Connect()
- if err == nil {
- if !m.running {
- go work(m)
- }
- m.running = true
- }
- return err
-}
-
-func (m *MessageCodec) Disconnect() error {
- log.Info().Msg("Disconnecting")
- m.running = false
- return m.transportInstance.Close()
-}
-
func (m *MessageCodec) Send(message interface{}) error {
log.Trace().Msg("Sending message")
// Cast the message to the correct type of struct
@@ -93,44 +53,19 @@ func (m *MessageCodec) Send(message interface{}) error {
}
// Send it to the PLC
- err = m.transportInstance.Write(wb.GetBytes())
+ err = m.TransportInstance.Write(wb.GetBytes())
if err != nil {
return errors.Wrap(err, "error sending request")
}
return nil
}
-func (m *MessageCodec) Expect(acceptsMessage spi.AcceptsMessage, handleMessage spi.HandleMessage, handleError spi.HandleError, ttl time.Duration) error {
- expectation := Expectation{
- expiration: time.Now().Add(ttl),
- acceptsMessage: acceptsMessage,
- handleMessage: handleMessage,
- handleError: handleError,
- }
- m.expectations = append(m.expectations, expectation)
- return nil
-}
-
-func (m *MessageCodec) SendRequest(message interface{}, acceptsMessage spi.AcceptsMessage, handleMessage spi.HandleMessage, handleError spi.HandleError, ttl time.Duration) error {
- log.Trace().Msg("Sending request")
- // Send the actual message
- err := m.Send(message)
- if err != nil {
- return errors.Wrap(err, "Error sending the request")
- }
- return m.Expect(acceptsMessage, handleMessage, handleError, ttl)
-}
-
-func (m *MessageCodec) GetDefaultIncomingMessageChannel() chan interface{} {
- return m.defaultIncomingMessageChannel
-}
-
-func (m *MessageCodec) receive() (interface{}, error) {
+func (m *MessageCodec) Receive() (interface{}, error) {
log.Trace().Msg("receiving")
// We need at least 6 bytes in order to know how big the packet is in total
- if num, err := m.transportInstance.GetNumReadableBytes(); (err == nil) && (num >= 6) {
+ if num, err := m.TransportInstance.GetNumReadableBytes(); (err == nil) && (num >= 6) {
log.Debug().Msgf("we got %d readable bytes", num)
- data, err := m.transportInstance.PeekReadableBytes(6)
+ data, err := m.TransportInstance.PeekReadableBytes(6)
if err != nil {
log.Warn().Err(err).Msg("error peeking")
// TODO: Possibly clean up ...
@@ -142,7 +77,7 @@ func (m *MessageCodec) receive() (interface{}, error) {
log.Debug().Msgf("Not enough bytes. Got: %d Need: %d\n", num, packetSize)
return nil, nil
}
- data, err = m.transportInstance.Read(packetSize)
+ data, err = m.TransportInstance.Read(packetSize)
if err != nil {
// TODO: Possibly clean up ...
return nil, nil
@@ -162,85 +97,3 @@ func (m *MessageCodec) receive() (interface{}, error) {
// TODO: maybe we return here a not enough error error
return nil, nil
}
-
-func work(m *MessageCodec) {
- // Start an endless loop
-mainLoop:
- for m.running {
- // Check for any expired expectations.
- // (Doing this outside the loop lets us expire expectations even if no input is coming in)
- now := time.Now()
-
- // Guard against empty expectations
- if len(m.expectations) <= 0 {
- log.Trace().Msg("we got no expectation")
- // Sleep for 10ms
- time.Sleep(time.Millisecond * 10)
- continue mainLoop
- }
- expectationLoop:
- for index, expectation := range m.expectations {
- // Check if this expectation has expired.
- if now.After(expectation.expiration) {
- // Remove this expectation from the list.
- m.expectations = append(m.expectations[:index], m.expectations[index+1:]...)
- // Call the error handler.
- err := expectation.handleError(plcerrors.NewTimeoutError(now.Sub(expectation.expiration)))
- if err != nil {
- log.Error().Err(err).Msg("Got an error handling error on expectation")
- }
- continue expectationLoop
- }
- }
-
- // Check for incoming messages.
- message, err := m.receive()
- if err != nil {
- log.Error().Err(err).Msg("got an error reading from transport")
- time.Sleep(time.Millisecond * 10)
- continue mainLoop
- }
- if message == nil {
- // Sleep for 10ms before checking again, in order to not
- // consume 100% CPU Power.
- time.Sleep(time.Millisecond * 10)
- continue mainLoop
- }
-
- // Go through all expectations
- messageHandled := false
- for index, expectation := range m.expectations {
- // Check if the current message matches the expectations
- // If it does, let it handle the message.
- if accepts := expectation.acceptsMessage(message); accepts {
- log.Debug().Stringer("expectation", expectation).Msg("accepts message")
- err = expectation.handleMessage(message)
- if err == nil {
- messageHandled = true
- // If this is the last element of the list remove it differently than if it's before that
- if (index + 1) == len(m.expectations) {
- m.expectations = m.expectations[:index]
- } else if (index + 1) < len(m.expectations) {
- m.expectations = append(m.expectations[:index], m.expectations[index+1:]...)
- }
- } else {
- // Pass the error to the error handler.
- err := expectation.handleError(err)
- if err != nil {
- log.Error().Err(err).Msg("Got an error handling error on expectation")
- }
- }
- }
- }
-
- // If the message has not been handled and a default handler is provided, call this ...
- if !messageHandled {
- // TODO: how do we prevent endless blocking if there is no reader on this channel?
- m.defaultIncomingMessageChannel <- message
- }
- }
-}
-
-func (m MessageCodec) GetTransportInstance() transports.TransportInstance {
- return m.transportInstance
-}
diff --git a/plc4go/internal/plc4go/spi/MessageCodec.go b/plc4go/internal/plc4go/spi/MessageCodec.go
index 65d74b4..2d69ecb 100644
--- a/plc4go/internal/plc4go/spi/MessageCodec.go
+++ b/plc4go/internal/plc4go/spi/MessageCodec.go
@@ -18,7 +18,22 @@
//
package spi
-import "time"
+import (
+ "fmt"
+ "github.com/apache/plc4x/plc4go/internal/plc4go/spi/plcerrors"
+ "github.com/apache/plc4x/plc4go/internal/plc4go/spi/transports"
+ "github.com/pkg/errors"
+ "github.com/rs/zerolog/log"
+ "time"
+)
+
+type Expectation interface {
+ GetExpiration() time.Time
+ GetAcceptsMessage() AcceptsMessage
+ GetHandleMessage() HandleMessage
+ GetHandleError() HandleError
+ fmt.Stringer
+}
// If this function returns true, the message is forwarded to the message handler
type AcceptsMessage func(message interface{}) bool
@@ -29,6 +44,33 @@ type HandleMessage func(message interface{}) error
// Function for handling the message, returns an error if anything goes wrong
type HandleError func(err error) error
+type DefaultExpectation struct {
+ Expiration time.Time
+ AcceptsMessage AcceptsMessage
+ HandleMessage HandleMessage
+ HandleError HandleError
+}
+
+func (m *DefaultExpectation) GetExpiration() time.Time {
+ return m.Expiration
+}
+
+func (m *DefaultExpectation) GetAcceptsMessage() AcceptsMessage {
+ return m.AcceptsMessage
+}
+
+func (m *DefaultExpectation) GetHandleMessage() HandleMessage {
+ return m.HandleMessage
+}
+
+func (m *DefaultExpectation) GetHandleError() HandleError {
+ return m.HandleError
+}
+
+func (m *DefaultExpectation) String() string {
+ return fmt.Sprintf("Expectation(expires at %v)", m.Expiration)
+}
+
type MessageCodec interface {
Connect() error
Disconnect() error
@@ -43,3 +85,179 @@ type MessageCodec interface {
GetDefaultIncomingMessageChannel() chan interface{}
}
+
+type DefaultCodecRequiredInterface interface {
+ MessageCodec
+ TimeoutExpectations(now time.Time)
+ HandleMessages(message interface{}) bool
+ Receive() (interface{}, error)
+}
+
+type DefaultCodec struct {
+ DefaultCodecRequiredInterface
+ TransportInstance transports.TransportInstance
+ DefaultIncomingMessageChannel chan interface{}
+ Expectations []Expectation
+ Running bool
+ CustomWorkLoop func(codec *DefaultCodecRequiredInterface)
+ CustomMessageHandling func(codec *DefaultCodecRequiredInterface, message interface{}) bool
+}
+
+func NewDefaultCodec(transportInstance transports.TransportInstance) *DefaultCodec {
+ return &DefaultCodec{
+ TransportInstance: transportInstance,
+ DefaultIncomingMessageChannel: make(chan interface{}),
+ Expectations: []Expectation{},
+ Running: false,
+ }
+}
+
+func (m *DefaultCodec) GetTransportInstance() transports.TransportInstance {
+ return m.TransportInstance
+}
+
+func (m *DefaultCodec) GetDefaultIncomingMessageChannel() chan interface{} {
+ return m.DefaultIncomingMessageChannel
+}
+
+func (m *DefaultCodec) Connect() error {
+ log.Info().Msg("Connecting")
+ err := m.TransportInstance.Connect()
+ if err == nil {
+ if !m.Running {
+ if m.CustomWorkLoop != nil {
+ go m.CustomWorkLoop(&m.DefaultCodecRequiredInterface)
+ } else {
+ go m.Work(&m.DefaultCodecRequiredInterface)
+ }
+
+ }
+ m.Running = true
+ }
+ return err
+}
+
+func (m *DefaultCodec) Disconnect() error {
+ log.Info().Msg("Disconnecting")
+ m.Running = false
+ return m.TransportInstance.Close()
+}
+
+func (m *DefaultCodec) Expect(acceptsMessage AcceptsMessage, handleMessage HandleMessage, handleError HandleError, ttl time.Duration) error {
+ expectation := &DefaultExpectation{
+ Expiration: time.Now().Add(ttl),
+ AcceptsMessage: acceptsMessage,
+ HandleMessage: handleMessage,
+ HandleError: handleError,
+ }
+ m.Expectations = append(m.Expectations, expectation)
+ return nil
+}
+
+func (m *DefaultCodec) SendRequest(message interface{}, acceptsMessage AcceptsMessage, handleMessage HandleMessage, handleError HandleError, ttl time.Duration) error {
+ log.Trace().Msg("Sending request")
+ // Send the actual message
+ err := m.Send(message)
+ if err != nil {
+ return errors.Wrap(err, "Error sending the request")
+ }
+ return m.Expect(acceptsMessage, handleMessage, handleError, ttl)
+}
+
+func (m *DefaultCodec) TimeoutExpectations(now time.Time) {
+ for index, expectation := range m.Expectations {
+ // Check if this expectation has expired.
+ if now.After(expectation.GetExpiration()) {
+ // Remove this expectation from the list.
+ m.Expectations = append(m.Expectations[:index], m.Expectations[index+1:]...)
+ // Call the error handler.
+ err := expectation.GetHandleError()(plcerrors.NewTimeoutError(now.Sub(expectation.GetExpiration())))
+ if err != nil {
+ log.Error().Err(err).Msg("Got an error handling error on expectation")
+ }
+ }
+ }
+}
+
+func (m *DefaultCodec) HandleMessages(message interface{}) bool {
+ messageHandled := false
+ for index, expectation := range m.Expectations {
+ // Check if the current message matches the expectations
+ // If it does, let it handle the message.
+ if accepts := expectation.GetAcceptsMessage()(message); accepts {
+ log.Debug().Stringer("expectation", expectation).Msg("accepts message")
+ err := expectation.GetHandleMessage()(message)
+ if err != nil {
+ // Pass the error to the error handler.
+ err := expectation.GetHandleError()(err)
+ if err != nil {
+ log.Error().Err(err).Msg("Got an error handling error on expectation")
+ }
+ continue
+ }
+ messageHandled = true
+ // If this is the last element of the list remove it differently than if it's before that
+ if (index + 1) == len(m.Expectations) {
+ m.Expectations = m.Expectations[:index]
+ } else if (index + 1) < len(m.Expectations) {
+ m.Expectations = append(m.Expectations[:index], m.Expectations[index+1:]...)
+ }
+ }
+ }
+ return messageHandled
+}
+
+func (m *DefaultCodec) Work(codec *DefaultCodecRequiredInterface) {
+ defer func() {
+ if err := recover(); err != nil {
+ log.Fatal().Msgf("recovered from %v", err)
+ }
+ log.Info().Msg("Keep running")
+ m.Work(codec)
+ }()
+ // Start an endless loop
+mainLoop:
+ for m.Running {
+ // Check for any expired expectations.
+ // (Doing this outside the loop lets us expire expectations even if no input is coming in)
+ now := time.Now()
+
+ // Guard against empty expectations
+ if len(m.Expectations) <= 0 {
+ log.Trace().Msg("we got no expectation")
+ // Sleep for 10ms
+ time.Sleep(time.Millisecond * 10)
+ continue mainLoop
+ }
+ m.TimeoutExpectations(now)
+
+ // Check for incoming messages.
+ message, err := m.Receive()
+ if err != nil {
+ log.Error().Err(err).Msg("got an error reading from transport")
+ time.Sleep(time.Millisecond * 10)
+ continue mainLoop
+ }
+ if message == nil {
+ // Sleep for 10ms before checking again, in order to not
+ // consume 100% CPU Power.
+ time.Sleep(time.Millisecond * 10)
+ continue mainLoop
+ }
+
+ if m.CustomMessageHandling != nil {
+ if m.CustomMessageHandling(codec, message) {
+ continue mainLoop
+ }
+ }
+
+ // Go through all expectations
+ messageHandled := m.HandleMessages(message)
+
+ // If the message has not been handled and a default handler is provided, call this ...
+ if !messageHandled {
+ // TODO: how do we prevent endless blocking if there is no reader on this channel?
+ m.DefaultIncomingMessageChannel <- message
+ }
+ }
+}