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/11/29 13:22:03 UTC
[plc4x] 03/03: feat(plc4go/bacnet): minor progress on bacnet
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 7949d53b4c2627eff11494528e3d1ad393de02a1
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Mon Nov 29 14:21:53 2021 +0100
feat(plc4go/bacnet): minor progress on bacnet
+ Try suppport for golang is still missing
---
.../drivers/tests/manual_bacnet_PcapTest_test.go | 15 ++++++++++++--
plc4go/internal/plc4go/bacnetip/Connection.go | 24 +++++++++++++++++++++-
plc4go/internal/plc4go/bacnetip/MessageCodec.go | 11 +++++++---
3 files changed, 44 insertions(+), 6 deletions(-)
diff --git a/plc4go/cmd/main/drivers/tests/manual_bacnet_PcapTest_test.go b/plc4go/cmd/main/drivers/tests/manual_bacnet_PcapTest_test.go
index ee7d378..137b70f 100644
--- a/plc4go/cmd/main/drivers/tests/manual_bacnet_PcapTest_test.go
+++ b/plc4go/cmd/main/drivers/tests/manual_bacnet_PcapTest_test.go
@@ -24,6 +24,8 @@ import (
"github.com/apache/plc4x/plc4go/internal/plc4go/spi"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi/transports/pcap"
"github.com/apache/plc4x/plc4go/pkg/plc4go"
+ "github.com/apache/plc4x/plc4go/pkg/plc4go/config"
+ "github.com/apache/plc4x/plc4go/pkg/plc4go/logging"
"github.com/apache/plc4x/plc4go/pkg/plc4go/model"
"github.com/rs/zerolog/log"
"io"
@@ -31,10 +33,16 @@ import (
"os"
"path"
"testing"
+ "time"
)
func Test(t *testing.T) {
t.Skip() // Manual test don't check in un-skipped
+
+ config.TraceTransactionManagerWorkers = false
+ config.TraceTransactionManagerTransactions = false
+ config.TraceDefaultMessageCodecWorker = false
+ logging.InfoLevel()
file := path.Join(os.TempDir(), "bacnet-stack-services.cap")
_, err := os.Stat(file)
if os.IsNotExist(err) {
@@ -46,7 +54,7 @@ func Test(t *testing.T) {
driverManager := plc4go.NewPlcDriverManager()
driverManager.RegisterDriver(bacnetip.NewDriver())
driverManager.(spi.TransportAware).RegisterTransport(pcap.NewTransport())
- result := <-driverManager.GetConnection("bacnet-ip:pcap://" + file + "?transport-type=udp")
+ result := <-driverManager.GetConnection("bacnet-ip:pcap://" + file + "?transport-type=udp&speed-factor=1")
if result.GetErr() != nil {
panic(result.GetErr())
}
@@ -67,7 +75,10 @@ func Test(t *testing.T) {
}
log.Info().Msgf("got response %v", requestResult.GetResponse())
- //time.Sleep(time.Hour)
+ for connection.IsConnected() {
+ log.Debug().Msg("Still sleeping")
+ time.Sleep(time.Second)
+ }
}
func download(dstPath string, url string) error {
diff --git a/plc4go/internal/plc4go/bacnetip/Connection.go b/plc4go/internal/plc4go/bacnetip/Connection.go
index 33434ee..b453327 100644
--- a/plc4go/internal/plc4go/bacnetip/Connection.go
+++ b/plc4go/internal/plc4go/bacnetip/Connection.go
@@ -46,6 +46,28 @@ func NewConnection(messageCodec spi.MessageCodec, fieldHandler spi.PlcFieldHandl
return connection
}
+func (c *Connection) Connect() <-chan plc4go.PlcConnectionConnectResult {
+ log.Trace().Msg("Connecting")
+ ch := make(chan plc4go.PlcConnectionConnectResult)
+ go func() {
+ connectionConnectResult := <-c.DefaultConnection.Connect()
+ go func() {
+ for c.IsConnected() {
+ log.Debug().Msg("Polling data")
+ incomingMessageChannel := c.messageCodec.GetDefaultIncomingMessageChannel()
+ select {
+ case message := <-incomingMessageChannel:
+ // TODO: implement mapping to subscribers
+ log.Info().Msgf("Received \n%v", message)
+ }
+ }
+ log.Info().Msg("Ending incoming message transfer")
+ }()
+ ch <- connectionConnectResult
+ }()
+ return ch
+}
+
func (c *Connection) GetConnection() plc4go.PlcConnection {
return c
}
@@ -59,7 +81,7 @@ func (c *Connection) SubscriptionRequestBuilder() model.PlcSubscriptionRequestBu
}
func (c *Connection) UnsubscriptionRequestBuilder() model.PlcUnsubscriptionRequestBuilder {
- panic("Not implementec yet. (at least as a default)")
+ panic("Not implemented yet. (at least as a default)")
}
func (c *Connection) addSubscriber(subscriber *Subscriber) {
diff --git a/plc4go/internal/plc4go/bacnetip/MessageCodec.go b/plc4go/internal/plc4go/bacnetip/MessageCodec.go
index 74d8d0e..14c75b9 100644
--- a/plc4go/internal/plc4go/bacnetip/MessageCodec.go
+++ b/plc4go/internal/plc4go/bacnetip/MessageCodec.go
@@ -35,7 +35,7 @@ type MessageCodec struct {
func NewMessageCodec(transportInstance transports.TransportInstance) *MessageCodec {
codec := &MessageCodec{}
- codec.DefaultCodec = _default.NewDefaultCodec(codec, transportInstance)
+ codec.DefaultCodec = _default.NewDefaultCodec(codec, transportInstance, _default.WithCustomMessageHandler(codec.handleCustomMessage))
return codec
}
@@ -73,8 +73,7 @@ func (m *MessageCodec) Receive() (interface{}, error) {
// TODO: Possibly clean up ...
return nil, nil
}
- //Second byte for the size and then add the header size 24
- packetSize := uint32((uint16(data[3]) << 8) + uint16(data[2]))
+ packetSize := uint32((uint16(data[2]) << 8) + uint16(data[3]))
if num < packetSize {
log.Debug().Msgf("Not enough bytes. Got: %d Need: %d\n", num, packetSize)
return nil, nil
@@ -100,3 +99,9 @@ func (m *MessageCodec) Receive() (interface{}, error) {
// TODO: maybe we return here a not enough error error
return nil, nil
}
+
+func (m *MessageCodec) handleCustomMessage(_ *_default.DefaultCodecRequirements, message interface{}) bool {
+ // For now, we just put them in the incoming channel
+ m.GetDefaultIncomingMessageChannel() <- message
+ return true
+}