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 2023/01/13 13:53:20 UTC
[plc4x] branch develop updated (5df3e9cf8f -> 609f6afbf2)
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 5df3e9cf8f build(deps): bump xmlunit.version from 2.9.0 to 2.9.1 (#740)
new 75872b46ef fix(plc4go/bacnet): smaller fixes relating udp comm
new 609f6afbf2 fix(plc4go/bacnet): fixed a bunch of broken code parts
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:
plc4go/internal/bacnetip/ApplicationLayer.go | 55 +++++---
plc4go/internal/bacnetip/ApplicationModule.go | 31 +++--
plc4go/internal/bacnetip/Core.go | 5 +-
plc4go/internal/bacnetip/IOCBModule.go | 16 ++-
plc4go/internal/bacnetip/MessageCodec.go | 33 +++--
plc4go/internal/bacnetip/NetworkService.go | 5 +-
plc4go/internal/bacnetip/PDU.go | 12 +-
plc4go/internal/bacnetip/Task.go | 150 +++++++++++++--------
.../internal/bacnetip/UDPCommunicationsModule.go | 8 +-
9 files changed, 196 insertions(+), 119 deletions(-)
[plc4x] 01/02: fix(plc4go/bacnet): smaller fixes relating udp comm
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 75872b46ef3d274d94d53265833e2e70d91ea76b
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Thu Jan 12 16:35:00 2023 +0100
fix(plc4go/bacnet): smaller fixes relating udp comm
---
plc4go/internal/bacnetip/ApplicationModule.go | 2 ++
plc4go/internal/bacnetip/MessageCodec.go | 5 ++++-
plc4go/internal/bacnetip/UDPCommunicationsModule.go | 5 ++++-
3 files changed, 10 insertions(+), 2 deletions(-)
diff --git a/plc4go/internal/bacnetip/ApplicationModule.go b/plc4go/internal/bacnetip/ApplicationModule.go
index 444ccc0a90..ac815d5358 100644
--- a/plc4go/internal/bacnetip/ApplicationModule.go
+++ b/plc4go/internal/bacnetip/ApplicationModule.go
@@ -448,6 +448,8 @@ func NewBIPSimpleApplication(localDevice *LocalDeviceObject, localAddress Addres
return nil, errors.Wrap(err, "error read write property services")
}
+ b.localAddress = localAddress
+
// include a application decoder
b.asap, err = NewApplicationServiceAccessPoint(nil, nil)
if err != nil {
diff --git a/plc4go/internal/bacnetip/MessageCodec.go b/plc4go/internal/bacnetip/MessageCodec.go
index ee70b515da..335cb46813 100644
--- a/plc4go/internal/bacnetip/MessageCodec.go
+++ b/plc4go/internal/bacnetip/MessageCodec.go
@@ -21,6 +21,7 @@ package bacnetip
import (
"context"
+ "fmt"
"github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model"
"github.com/apache/plc4x/plc4go/spi"
"github.com/apache/plc4x/plc4go/spi/default"
@@ -49,7 +50,6 @@ func NewApplicationLayerMessageCodec(udpTransport *udp.Transport, transportUrl u
if err != nil {
return nil, errors.Wrap(err, "error creating transport instance")
}
- _ = transportInstance
a := &ApplicationLayerMessageCodec{
localAddress: localAddress,
remoteAddress: remoteAddress,
@@ -58,6 +58,9 @@ func NewApplicationLayerMessageCodec(udpTransport *udp.Transport, transportUrl u
if err != nil {
return nil, errors.Wrap(err, "error creating address")
}
+ // TODO: workaround for strange address parsing
+ at := AddressTuple[string, uint16]{fmt.Sprintf("%d.%d.%d.%d", address.AddrAddress[0], address.AddrAddress[1], address.AddrAddress[2], address.AddrAddress[3]), *address.AddrPort}
+ address.AddrTuple = &at
application, err := NewBIPSimpleApplication(&LocalDeviceObject{}, *address, &a.deviceInfoCache, nil)
if err != nil {
return nil, err
diff --git a/plc4go/internal/bacnetip/UDPCommunicationsModule.go b/plc4go/internal/bacnetip/UDPCommunicationsModule.go
index b29bfbb994..8d5984f369 100644
--- a/plc4go/internal/bacnetip/UDPCommunicationsModule.go
+++ b/plc4go/internal/bacnetip/UDPCommunicationsModule.go
@@ -149,6 +149,9 @@ func NewUDPDirector(address AddressTuple[string, uint16], timeout *int, reuse *b
return nil, errors.Wrap(err, "error resolving udp address")
}
d.ti = udp.NewTransportInstance(resolvedAddress, nil, d.timeout, d.reuse, nil)
+ if err := d.ti.Connect(); err != nil {
+ return nil, errors.Wrap(err, "error connecting transport instance")
+ }
d.running = true
go func() {
@@ -164,7 +167,7 @@ func NewUDPDirector(address AddressTuple[string, uint16], timeout *int, reuse *b
}()
// start with an empty peer pool
- d.peers = nil
+ d.peers = map[string]*UDPActor{}
return d, nil
}
[plc4x] 02/02: fix(plc4go/bacnet): fixed a bunch of broken code parts
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 609f6afbf21e45e713392ba0c475c98de3a7228d
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Fri Jan 13 14:52:57 2023 +0100
fix(plc4go/bacnet): fixed a bunch of broken code parts
---
plc4go/internal/bacnetip/ApplicationLayer.go | 55 +++++---
plc4go/internal/bacnetip/ApplicationModule.go | 29 ++--
plc4go/internal/bacnetip/Core.go | 5 +-
plc4go/internal/bacnetip/IOCBModule.go | 16 ++-
plc4go/internal/bacnetip/MessageCodec.go | 28 ++--
plc4go/internal/bacnetip/NetworkService.go | 5 +-
plc4go/internal/bacnetip/PDU.go | 12 +-
plc4go/internal/bacnetip/Task.go | 150 +++++++++++++--------
.../internal/bacnetip/UDPCommunicationsModule.go | 3 +-
9 files changed, 186 insertions(+), 117 deletions(-)
diff --git a/plc4go/internal/bacnetip/ApplicationLayer.go b/plc4go/internal/bacnetip/ApplicationLayer.go
index 6fc4a2bf2a..1dcaf01551 100644
--- a/plc4go/internal/bacnetip/ApplicationLayer.go
+++ b/plc4go/internal/bacnetip/ApplicationLayer.go
@@ -89,9 +89,14 @@ type SSMSAPRequirements interface {
GetDefaultMaximumApduLengthAccepted() readWriteModel.MaxApduLengthAccepted
}
+type SSMProcessingRequirements interface {
+ processTask() error
+}
+
// SSM - Segmentation State Machine
type SSM struct {
*OneShotTask
+ SSMProcessingRequirements
ssmSAP SSMSAPRequirements
@@ -120,7 +125,10 @@ type SSM struct {
maxApduLengthAccepted readWriteModel.MaxApduLengthAccepted
}
-func NewSSM(sap SSMSAPRequirements, pduAddress *Address) (SSM, error) {
+func NewSSM(sap interface {
+ SSMSAPRequirements
+ SSMProcessingRequirements
+}, pduAddress *Address) (*SSM, error) {
log.Debug().Interface("sap", sap).Interface("pdu_address", pduAddress).Msg("init")
var deviceInfo *DeviceInfo
deviceInfoTemp, ok := sap.GetDeviceInfoCache().GetDeviceInfo(DeviceInfoCacheKey{PduSource: pduAddress})
@@ -162,8 +170,7 @@ func NewSSM(sap SSMSAPRequirements, pduAddress *Address) (SSM, error) {
} else {
maxApduLengthAccepted = sap.GetDefaultMaximumApduLengthAccepted()
}
- return SSM{
- OneShotTask: NewOneShotTask(nil),
+ ssm := &SSM{
ssmSAP: sap,
pduAddress: pduAddress,
deviceInfo: deviceInfo,
@@ -174,7 +181,10 @@ func NewSSM(sap SSMSAPRequirements, pduAddress *Address) (SSM, error) {
segmentTimeout: segmentTimeout,
maxSegmentsAccepted: maxSegmentsAccepted,
maxApduLengthAccepted: maxApduLengthAccepted,
- }, nil
+ }
+ ssm.OneShotTask = NewOneShotTask(ssm, nil)
+ ssm.SSMProcessingRequirements = sap
+ return ssm, nil
}
func (s *SSM) StartTimer(millis uint) {
@@ -363,7 +373,7 @@ func (s *SSM) fillWindow(sequenceNumber uint8) error {
if err != nil {
return errors.Wrapf(err, "Error sending out segment %d", i)
}
- if err := s.ssmSAP.Request(NewPDU(apdu, WithPDUDestination(s.pduAddress))); err != nil {
+ if err := s.ssmSAP.Request(NewPDU(apdu.GetMessage(), WithPDUDestination(s.pduAddress))); err != nil {
log.Debug().Err(err).Msg("error sending request")
}
if moreFollows {
@@ -374,12 +384,16 @@ func (s *SSM) fillWindow(sequenceNumber uint8) error {
}
type ClientSSM struct {
- SSM
+ *SSM
}
func NewClientSSM(sap SSMSAPRequirements, pduAddress *Address) (*ClientSSM, error) {
log.Debug().Interface("sap", sap).Interface("pduAddress", pduAddress).Msg("init")
- ssm, err := NewSSM(sap, pduAddress)
+ c := &ClientSSM{}
+ ssm, err := NewSSM(struct {
+ SSMSAPRequirements
+ SSMProcessingRequirements
+ }{sap, c}, pduAddress)
if err != nil {
return nil, err
}
@@ -388,14 +402,13 @@ func NewClientSSM(sap SSMSAPRequirements, pduAddress *Address) (*ClientSSM, erro
// TODO: get entry for device, store it in inventory
log.Debug().Msg("Accquire device information")
}
- return &ClientSSM{
- SSM: ssm,
- }, nil
+ c.SSM = ssm
+ return c, nil
}
// setState This function is called when the client wants to change state
func (c *ClientSSM) setState(newState SSMState, timer *uint) error {
- log.Debug().Msgf("setState %c timer=%d", newState, timer)
+ log.Debug().Msgf("setState %s timer=%d", newState, timer)
// do the regular state change
if err := c.SSM.setState(newState, timer); err != nil {
return errors.Wrap(err, "error during SSM state transition")
@@ -556,7 +569,7 @@ func (c *ClientSSM) Confirmation(apdu _PDU) error {
// processTask This function is called when something has taken too long
func (c *ClientSSM) processTask() error {
- log.Debug().Msg("processTask")
+ log.Debug().Msgf("processTask (currentState: %s)", c.state)
switch c.state {
case SSMState_SEGMENTED_REQUEST:
return c.segmentedRequestTimeout()
@@ -826,7 +839,7 @@ func (c *ClientSSM) awaitConfirmationTimeout() error {
}
c.retryCount = saveCount
} else {
- log.Debug().Msg("retry count exceeded")
+ log.Debug().Msgf("retry count exceeded: %d >= %d", c.retryCount, c.numberOfApduRetries)
abort, err := c.abort(readWriteModel.BACnetAbortReason(65)) // Note: this is a proprietary code used by bacpypes for no response. We just use that here too to keep consistent
if err != nil {
@@ -945,13 +958,19 @@ func (c *ClientSSM) segmentedConfirmationTimeout() error {
}
type ServerSSM struct {
- SSM
+ *SSM
segmentedResponseAccepted bool
}
func NewServerSSM(sap SSMSAPRequirements, pduAddress *Address) (*ServerSSM, error) {
log.Debug().Interface("sap", sap).Interface("pduAddress", pduAddress).Msg("init")
- ssm, err := NewSSM(sap, pduAddress)
+ s := &ServerSSM{
+ segmentedResponseAccepted: true,
+ }
+ ssm, err := NewSSM(struct {
+ SSMSAPRequirements
+ SSMProcessingRequirements
+ }{sap, s}, pduAddress)
if err != nil {
return nil, err
}
@@ -960,10 +979,8 @@ func NewServerSSM(sap SSMSAPRequirements, pduAddress *Address) (*ServerSSM, erro
// TODO: get entry for device, store it in inventory
log.Debug().Msg("Accquire device information")
}
- return &ServerSSM{
- SSM: ssm,
- segmentedResponseAccepted: true,
- }, nil
+ s.SSM = ssm
+ return s, nil
}
// setState This function is called when the client wants to change state
diff --git a/plc4go/internal/bacnetip/ApplicationModule.go b/plc4go/internal/bacnetip/ApplicationModule.go
index ac815d5358..5d0cd114b9 100644
--- a/plc4go/internal/bacnetip/ApplicationModule.go
+++ b/plc4go/internal/bacnetip/ApplicationModule.go
@@ -26,7 +26,6 @@ import (
"github.com/pkg/errors"
"github.com/rs/zerolog/log"
"hash/fnv"
- "net"
)
type DeviceInfo struct {
@@ -214,7 +213,7 @@ type Application struct {
controllers map[string]interface{}
helpers map[string]func(pdu _PDU) error
- _startup_disabled bool
+ _startupDisabled bool
}
func NewApplication(localDevice *LocalDeviceObject, localAddress Address, deviceInfoCache *DeviceInfoCache, aseID *int) (*Application, error) {
@@ -256,7 +255,7 @@ func NewApplication(localDevice *LocalDeviceObject, localAddress Address, device
a.Collector = Collector{}
// if starting up is enabled, find all the startup functions
- if !a._startup_disabled {
+ if !a._startupDisabled {
for _, fn := range a.CapabilityFunctions("startup") {
log.Debug().Msgf("startup fn %t", fn != nil)
fn()
@@ -296,7 +295,9 @@ func (a *Application) Indication(apdu _PDU) error {
if err := helperFn(apdu); err != nil {
log.Debug().Err(err).Msgf("err result")
// TODO: do proper mapping
- a.Response(NewPDU(readWriteModel.NewAPDUError(0, readWriteModel.BACnetConfirmedServiceChoice_CREATE_OBJECT, nil, 0)))
+ if err := a.Response(NewPDU(readWriteModel.NewAPDUError(0, readWriteModel.BACnetConfirmedServiceChoice_CREATE_OBJECT, nil, 0))); err != nil {
+ return err
+ }
}
return nil
@@ -346,7 +347,7 @@ func (a *ApplicationIOController) ProcessIO(iocb _IOCB) error {
return queue.RequestIO(iocb)
}
-func (a *ApplicationIOController) _AppComplete(address net.Addr, apdu _PDU) error {
+func (a *ApplicationIOController) _AppComplete(address *Address, apdu _PDU) error {
log.Debug().Msgf("_AppComplete %s\n%s", address, apdu)
// look up the queue
@@ -366,10 +367,14 @@ func (a *ApplicationIOController) _AppComplete(address net.Addr, apdu _PDU) erro
// this request is complete
switch apdu.(type) {
case readWriteModel.APDUSimpleAckExactly, readWriteModel.APDUComplexAckExactly:
- queue.CompleteIO(queue.activeIOCB, apdu)
+ if err := queue.CompleteIO(queue.activeIOCB, apdu); err != nil {
+ return err
+ }
case readWriteModel.APDUErrorExactly, readWriteModel.APDURejectExactly, readWriteModel.APDUAbortExactly:
// TODO: extract error
- queue.AbortIO(queue.activeIOCB, errors.Errorf("%s", apdu))
+ if err := queue.AbortIO(queue.activeIOCB, errors.Errorf("%s", apdu)); err != nil {
+ return err
+ }
default:
return errors.New("unrecognized APDU type")
}
@@ -392,8 +397,10 @@ func (a *ApplicationIOController) _AppRequest(apdu _PDU) {
// if this was an unconfirmed request, it's complete, no message
if _, ok := apdu.(readWriteModel.APDUUnconfirmedRequestExactly); ok {
- // TODO: where to get the destination now again??
- a._AppComplete(nil, apdu)
+ if err := a._AppComplete(apdu.GetPDUDestination(), apdu); err != nil {
+ log.Error().Err(err).Msg("AppRequest failed")
+ return
+ }
}
}
@@ -413,9 +420,7 @@ func (a *ApplicationIOController) Confirmation(apdu _PDU) error {
log.Debug().Msgf("Confirmation\n%s", apdu)
// this is an ack, error, reject or abort
- // TODO: where to get the destination now again??
- a._AppComplete(nil, apdu)
- return nil
+ return a._AppComplete(apdu.GetPDUDestination(), apdu)
}
type BIPSimpleApplication struct {
diff --git a/plc4go/internal/bacnetip/Core.go b/plc4go/internal/bacnetip/Core.go
index 44b508fe20..145617be80 100644
--- a/plc4go/internal/bacnetip/Core.go
+++ b/plc4go/internal/bacnetip/Core.go
@@ -19,6 +19,9 @@
package bacnetip
-func Deferred(fn func()) {
+import "github.com/rs/zerolog/log"
+
+func Deferred(fn func() error) {
+ log.Debug().Msg("Deferred")
// TODO: implement me
}
diff --git a/plc4go/internal/bacnetip/IOCBModule.go b/plc4go/internal/bacnetip/IOCBModule.go
index d7dfaf37c1..a886eb212b 100644
--- a/plc4go/internal/bacnetip/IOCBModule.go
+++ b/plc4go/internal/bacnetip/IOCBModule.go
@@ -741,19 +741,19 @@ func (i *IOQController) AbortIO(iocb _IOCB, err error) error {
}
// _trigger Called to launch the next request in the queue
-func (i *IOQController) _trigger() {
+func (i *IOQController) _trigger() error {
log.Debug().Msg("_trigger")
// if we are busy, do nothing
if i.state != IOQControllerStates_CTRL_IDLE {
log.Debug().Msg("not idle")
- return
+ return nil
}
// if there is nothing to do, return
if len(i.ioQueue.queue) == 0 {
log.Debug().Msg("empty queue")
- return
+ return nil
}
// get the next iocb
@@ -766,25 +766,26 @@ func (i *IOQController) _trigger() {
// if there was an error, abort the request
if err := i.Abort(err); err != nil {
log.Debug().Err(err).Msg("error aborting")
- return
+ return nil
}
- return
+ return nil
}
// if we're idle, call again
if i.state == IOQControllerStates_CTRL_IDLE {
Deferred(i._trigger)
}
+ return nil
}
// _waitTrigger is called to launch the next request in the queue
-func (i *IOQController) _waitTrigger() {
+func (i *IOQController) _waitTrigger() error {
log.Debug().Msg("_waitTrigger")
// make sure we are waiting
if i.state != IOQControllerStates_CTRL_WAITING {
log.Debug().Msg("not waiting")
- return
+ return nil
}
// change our state
@@ -793,6 +794,7 @@ func (i *IOQController) _waitTrigger() {
// look for more to do
i._trigger()
+ return nil
}
type SieveQueue struct {
diff --git a/plc4go/internal/bacnetip/MessageCodec.go b/plc4go/internal/bacnetip/MessageCodec.go
index 335cb46813..5fe764b7dd 100644
--- a/plc4go/internal/bacnetip/MessageCodec.go
+++ b/plc4go/internal/bacnetip/MessageCodec.go
@@ -45,11 +45,12 @@ type ApplicationLayerMessageCodec struct {
}
func NewApplicationLayerMessageCodec(udpTransport *udp.Transport, transportUrl url.URL, options map[string][]string, localAddress *net.UDPAddr, remoteAddress *net.UDPAddr) (*ApplicationLayerMessageCodec, error) {
+ // TODO: currently this is done by the BIP down below
// Have the transport create a new transport-instance.
- transportInstance, err := udpTransport.CreateTransportInstanceForLocalAddress(transportUrl, options, localAddress)
- if err != nil {
- return nil, errors.Wrap(err, "error creating transport instance")
- }
+ //transportInstance, err := udpTransport.CreateTransportInstanceForLocalAddress(transportUrl, options, localAddress)
+ //if err != nil {
+ // return nil, errors.Wrap(err, "error creating transport instance")
+ //}
a := &ApplicationLayerMessageCodec{
localAddress: localAddress,
remoteAddress: remoteAddress,
@@ -61,12 +62,15 @@ func NewApplicationLayerMessageCodec(udpTransport *udp.Transport, transportUrl u
// TODO: workaround for strange address parsing
at := AddressTuple[string, uint16]{fmt.Sprintf("%d.%d.%d.%d", address.AddrAddress[0], address.AddrAddress[1], address.AddrAddress[2], address.AddrAddress[3]), *address.AddrPort}
address.AddrTuple = &at
- application, err := NewBIPSimpleApplication(&LocalDeviceObject{}, *address, &a.deviceInfoCache, nil)
+ application, err := NewBIPSimpleApplication(&LocalDeviceObject{
+ NumberOfAPDURetries: func() *uint { retries := uint(10); return &retries }(),
+ }, *address, &a.deviceInfoCache, nil)
if err != nil {
return nil, err
}
a.bipSimpleApplication = application
- a.messageCode = NewMessageCodec(transportInstance)
+ // TODO: this is currently done by the BIP
+ //a.messageCode = NewMessageCodec(transportInstance)
return a, nil
}
@@ -75,11 +79,15 @@ func (m *ApplicationLayerMessageCodec) GetCodec() spi.MessageCodec {
}
func (m *ApplicationLayerMessageCodec) Connect() error {
- return m.messageCode.Connect()
+ // TODO: this is currently done by the BIP
+ // return m.messageCode.Connect()
+ return nil
}
func (m *ApplicationLayerMessageCodec) ConnectWithContext(ctx context.Context) error {
- return m.messageCode.ConnectWithContext(ctx)
+ // TODO: this is currently done by the BIP
+ // return m.messageCode.ConnectWithContext(ctx)
+ return nil
}
func (m *ApplicationLayerMessageCodec) Disconnect() error {
@@ -132,7 +140,9 @@ func (m *ApplicationLayerMessageCodec) SendRequest(ctx context.Context, message
}
func (m *ApplicationLayerMessageCodec) GetDefaultIncomingMessageChannel() chan spi.Message {
- return m.messageCode.GetDefaultIncomingMessageChannel()
+ // TODO: this is currently done by the BIP
+ //return m.messageCode.GetDefaultIncomingMessageChannel()
+ return make(chan spi.Message)
}
type MessageCodec struct {
diff --git a/plc4go/internal/bacnetip/NetworkService.go b/plc4go/internal/bacnetip/NetworkService.go
index c00ff11c04..4992235732 100644
--- a/plc4go/internal/bacnetip/NetworkService.go
+++ b/plc4go/internal/bacnetip/NetworkService.go
@@ -568,13 +568,14 @@ func NewNetworkServiceElement(eid *int) (*NetworkServiceElement, error) {
return n, nil
}
-func (n *NetworkServiceElement) Startup() {
+func (n *NetworkServiceElement) Startup() error {
log.Debug().Msg("Startup")
// reference the service access point
- sap := n.elementService.(*NetworkServiceAccessPoint) // TODO: hard cast but seems like adapters apears first in network service access point (so hard binding)
+ sap := n.elementService.(*NetworkServiceAccessPoint) // TODO: hard cast but seems like adapters appears first in network service access point (so hard binding)
log.Debug().Msgf("sap: %v", sap)
// loop through all the adapters
// TODO: no adapters yet
+ return nil
}
diff --git a/plc4go/internal/bacnetip/PDU.go b/plc4go/internal/bacnetip/PDU.go
index a468c9caf4..35fe6b18b8 100644
--- a/plc4go/internal/bacnetip/PDU.go
+++ b/plc4go/internal/bacnetip/PDU.go
@@ -423,7 +423,6 @@ func NewPCI(msg spi.Message, pduSource *Address, pduDestination *Address, expect
}
type _PDU interface {
- spi.Message
GetMessage() spi.Message
GetPDUSource() *Address
GetPDUDestination() *Address
@@ -433,13 +432,11 @@ type _PDU interface {
}
type PDU struct {
- spi.Message
*PCI
}
func NewPDU(msg spi.Message, pduOptions ...PDUOption) *PDU {
p := &PDU{
- msg,
NewPCI(msg, nil, nil, false, readWriteModel.NPDUNetworkPriority_NORMAL_MESSAGE),
}
for _, option := range pduOptions {
@@ -449,9 +446,8 @@ func NewPDU(msg spi.Message, pduOptions ...PDUOption) *PDU {
}
func NewPDUFromPDU(pdu _PDU, pduOptions ...PDUOption) *PDU {
- msg := pdu.(*PDU).Message
+ msg := pdu.(*PDU).pduUserData
p := &PDU{
- msg,
NewPCI(msg, pdu.GetPDUSource(), pdu.GetPDUDestination(), pdu.GetExpectingReply(), pdu.GetNetworkPriority()),
}
for _, option := range pduOptions {
@@ -462,7 +458,6 @@ func NewPDUFromPDU(pdu _PDU, pduOptions ...PDUOption) *PDU {
func NewPDUFromPDUWithNewMessage(pdu _PDU, msg spi.Message, pduOptions ...PDUOption) *PDU {
p := &PDU{
- msg,
NewPCI(msg, pdu.GetPDUSource(), pdu.GetPDUDestination(), pdu.GetExpectingReply(), pdu.GetNetworkPriority()),
}
for _, option := range pduOptions {
@@ -473,7 +468,6 @@ func NewPDUFromPDUWithNewMessage(pdu _PDU, msg spi.Message, pduOptions ...PDUOpt
func NewPDUWithAllOptions(msg spi.Message, pduSource *Address, pduDestination *Address, expectingReply bool, networkPriority readWriteModel.NPDUNetworkPriority) *PDU {
return &PDU{
- msg,
NewPCI(msg, pduSource, pduDestination, expectingReply, networkPriority),
}
}
@@ -505,7 +499,7 @@ func WithPDUNetworkPriority(networkPriority readWriteModel.NPDUNetworkPriority)
}
func (p *PDU) GetMessage() spi.Message {
- return p.Message
+ return p.pduUserData
}
func (p *PDU) GetPDUSource() *Address {
@@ -529,5 +523,5 @@ func (p *PDU) GetNetworkPriority() readWriteModel.NPDUNetworkPriority {
}
func (p *PDU) String() string {
- return fmt.Sprintf("PDU{\n%s,\n%s}", p.Message, p._PCI)
+ return fmt.Sprintf("PDU{\n%s}", p._PCI)
}
diff --git a/plc4go/internal/bacnetip/Task.go b/plc4go/internal/bacnetip/Task.go
index f2ea602325..5e0a84ec11 100644
--- a/plc4go/internal/bacnetip/Task.go
+++ b/plc4go/internal/bacnetip/Task.go
@@ -26,17 +26,21 @@ import (
)
type _TaskRequirements interface {
- processTask()
+ processTask() error
+ InstallTask(when *time.Time, delta *time.Duration)
+ getTaskTime() *time.Time
+ getIsScheduled() bool
+ setIsScheduled(isScheduled bool)
}
type _Task struct {
- _TaskRequirements
- taskTime *time.Time
- isScheduled bool
+ taskRequirements _TaskRequirements
+ taskTime *time.Time
+ isScheduled bool
}
func _New_Task(_TaskRequirements _TaskRequirements) *_Task {
- return &_Task{_TaskRequirements: _TaskRequirements}
+ return &_Task{taskRequirements: _TaskRequirements}
}
func (t *_Task) InstallTask(when *time.Time, delta *time.Duration) {
@@ -57,27 +61,42 @@ func (t *_Task) InstallTask(when *time.Time, delta *time.Duration) {
t.taskTime = when
// pass along to the task manager
- _taskManager.installTask(t)
-}
-
-func (t *_Task) processTask() {
- t._TaskRequirements.processTask()
+ _taskManager.installTask(t.taskRequirements)
}
func (t *_Task) SuspendTask() {
- _taskManager.suspendTask(t)
+ _taskManager.suspendTask(t.taskRequirements)
}
func (t *_Task) Resume() {
- _taskManager.resumeTask(t)
+ _taskManager.resumeTask(t.taskRequirements)
+}
+
+func (t *_Task) getTaskTime() *time.Time {
+ return t.taskTime
+}
+
+func (t *_Task) getIsScheduled() bool {
+ return t.isScheduled
+}
+
+func (t *_Task) setIsScheduled(isScheduled bool) {
+ t.isScheduled = isScheduled
+}
+
+type OneShotTaskRequirements interface {
+ processTask() error
}
type OneShotTask struct {
*_Task
+ OneShotTaskRequirements
}
-func NewOneShotTask(when *time.Time) *OneShotTask {
- o := &OneShotTask{}
+func NewOneShotTask(oneShotTaskRequirements OneShotTaskRequirements, when *time.Time) *OneShotTask {
+ o := &OneShotTask{
+ OneShotTaskRequirements: oneShotTaskRequirements,
+ }
o._Task = _New_Task(o)
if when != nil {
o.taskTime = when
@@ -87,10 +106,11 @@ func NewOneShotTask(when *time.Time) *OneShotTask {
type OneShotDeleteTask struct {
*_Task
+ OneShotTaskRequirements
}
-func NewOneShotDeleteTask(when *time.Time) *OneShotDeleteTask {
- o := &OneShotDeleteTask{}
+func NewOneShotDeleteTask(oneShotTaskRequirements OneShotTaskRequirements, when *time.Time) *OneShotDeleteTask {
+ o := &OneShotDeleteTask{OneShotTaskRequirements: oneShotTaskRequirements}
o._Task = _New_Task(o)
if when != nil {
o.taskTime = when
@@ -98,39 +118,49 @@ func NewOneShotDeleteTask(when *time.Time) *OneShotDeleteTask {
return o
}
+func (r *OneShotDeleteTask) IsOneShotDeleteTask() bool {
+ return true
+}
+
type OneShotFunctionTask struct {
*OneShotDeleteTask
- fn func()
+ fn func() error
}
-func (m *OneShotFunctionTask) processTask() {
- m.fn()
+func (m *OneShotFunctionTask) processTask() error {
+ return m.fn()
}
-func OneShotFunction(fn func()) *OneShotFunctionTask {
- task := &OneShotFunctionTask{NewOneShotDeleteTask(nil), fn}
-
+func OneShotFunction(fn func() error) *OneShotFunctionTask {
+ task := &OneShotFunctionTask{fn: fn}
+ task.OneShotDeleteTask = NewOneShotDeleteTask(task, nil)
var delta time.Duration = 0
task.InstallTask(nil, &delta)
return task
}
-func FunctionTask(fn func()) *OneShotFunctionTask {
- task := &OneShotFunctionTask{NewOneShotDeleteTask(nil), fn}
+func FunctionTask(fn func() error) *OneShotFunctionTask {
+ task := &OneShotFunctionTask{fn: fn}
+ task.OneShotDeleteTask = NewOneShotDeleteTask(task, nil)
log.Debug().Msgf("task: %v", task)
return task
}
+type RecurringTaskRequirements interface {
+ processTask() error
+}
+
type RecurringTask struct {
*_Task
+ RecurringTaskRequirements
taskInterval *time.Duration
taskIntervalOffset *time.Duration
}
-func NewRecurringTask(_TaskRequirements _TaskRequirements, interval *time.Duration, offset *time.Duration) *RecurringTask {
- r := &RecurringTask{}
- r._Task = _New_Task(_TaskRequirements)
+func NewRecurringTask(recurringTaskRequirements RecurringTaskRequirements, interval *time.Duration, offset *time.Duration) *RecurringTask {
+ r := &RecurringTask{RecurringTaskRequirements: recurringTaskRequirements}
+ r._Task = _New_Task(r)
// set the interval if it hasn't already been set
if interval != nil {
r.taskInterval = interval
@@ -163,28 +193,32 @@ func NewRecurringTask(_TaskRequirements _TaskRequirements, interval *time.Durati
r.taskTime = &_taskTime
// install it
- _taskManager.installTask(r._Task)
+ _taskManager.installTask(r)
return r
}
+func (r *RecurringTask) IsRecurringTask() bool {
+ return true
+}
+
type _RecurringFunctionTask struct {
*RecurringTask
- fn func()
+ fn func() error
}
-func _New_RecurringFunctionTask(interval *time.Duration) *_RecurringFunctionTask {
- r := &_RecurringFunctionTask{}
+func _New_RecurringFunctionTask(interval *time.Duration, fn func() error) *_RecurringFunctionTask {
+ r := &_RecurringFunctionTask{fn: fn}
r.RecurringTask = NewRecurringTask(r, interval, nil)
return r
}
-func (r _RecurringFunctionTask) processTask() {
- r.fn()
+func (r _RecurringFunctionTask) processTask() error {
+ return r.fn()
}
-func RecurringFunctionTask(interval *time.Duration) *RecurringTask {
- return _New_RecurringFunctionTask(interval).RecurringTask
+func RecurringFunctionTask(interval *time.Duration, fn func() error) *RecurringTask {
+ return _New_RecurringFunctionTask(interval, fn).RecurringTask
}
var _taskManager = TaskManager{}
@@ -205,25 +239,25 @@ func init() {
type TaskManager struct {
sync.Mutex
- tasks []*_Task
+ tasks []_TaskRequirements
}
func (m *TaskManager) getTime() time.Time {
return time.Now()
}
-func (m *TaskManager) installTask(task *_Task) {
+func (m *TaskManager) installTask(task _TaskRequirements) {
m.Lock()
defer m.Unlock()
- log.Debug().Msgf("installTask %v@%v", task, task.taskTime)
+ log.Debug().Msgf("installTask %v@%v", task, task.getTaskTime())
// if the taskTime is None is hasn't been computed correctly
- if task.taskTime == nil {
+ if task.getTaskTime() == nil {
panic("task time is None")
}
// if this is already installed, suspend it
- if task.isScheduled {
+ if task.getIsScheduled() {
m.suspendTask(task)
}
@@ -231,10 +265,10 @@ func (m *TaskManager) installTask(task *_Task) {
// TODO: we might need to insert it at the right place
m.tasks = append(m.tasks, task)
- task.isScheduled = true
+ task.setIsScheduled(true)
}
-func (m *TaskManager) suspendTask(task *_Task) {
+func (m *TaskManager) suspendTask(task _TaskRequirements) {
log.Debug().Msgf("suspendTask %v", task)
m.Lock()
defer m.Unlock()
@@ -244,7 +278,7 @@ func (m *TaskManager) suspendTask(task *_Task) {
if _task == task {
log.Debug().Msgf("task found")
iToDelete = i
- task.isScheduled = false
+ task.setIsScheduled(false)
break
}
}
@@ -255,7 +289,7 @@ func (m *TaskManager) suspendTask(task *_Task) {
}
}
-func (m *TaskManager) resumeTask(task *_Task) {
+func (m *TaskManager) resumeTask(task _TaskRequirements) {
log.Debug().Msgf("resumeTask %v", task)
m.Lock()
defer m.Unlock()
@@ -264,28 +298,28 @@ func (m *TaskManager) resumeTask(task *_Task) {
m.installTask(task)
}
-func (m *TaskManager) getNextTask() (*_Task, time.Duration) {
- log.Debug().Msgf("getNextTask")
+func (m *TaskManager) getNextTask() (_TaskRequirements, time.Duration) {
+ log.Trace().Msgf("getNextTask")
m.Lock()
defer m.Unlock()
now := time.Now()
- var task *_Task
+ var task _TaskRequirements
var delta time.Duration
- if m.tasks != nil {
+ if len(m.tasks) > 0 {
nextTask := m.tasks[0]
- when := nextTask.taskTime
+ when := nextTask.getTaskTime()
if when.Before(now) {
// pull it off the list and mark that it's no longer scheduled
m.tasks = m.tasks[1:] // TODO: guard against empty list
task = nextTask
- task.isScheduled = false
+ task.setIsScheduled(false)
- if m.tasks != nil {
+ if len(m.tasks) > 0 {
nextTask = m.tasks[0]
- when = nextTask.taskTime
+ when = nextTask.getTaskTime()
// peek at the next task, return how long to wait
delta = when.Sub(now) // TODO: avoid negative
}
@@ -298,16 +332,18 @@ func (m *TaskManager) getNextTask() (*_Task, time.Duration) {
return task, delta
}
-func (m *TaskManager) processTask(task *_Task) {
+func (m *TaskManager) processTask(task _TaskRequirements) {
log.Debug().Msgf("processTask %v", task)
// process the task
- task.processTask()
+ if err := task.processTask(); err != nil {
+ log.Error().Err(err).Msg("Error processing Task")
+ }
- switch task._TaskRequirements.(type) {
- case *RecurringTask:
+ switch task.(type) {
+ case interface{ IsRecurringTask() bool }:
task.InstallTask(nil, nil)
- case OneShotDeleteTask:
+ case interface{ IsOneShotDeleteTask() bool }:
// TODO: Delete? How?
}
}
diff --git a/plc4go/internal/bacnetip/UDPCommunicationsModule.go b/plc4go/internal/bacnetip/UDPCommunicationsModule.go
index 8d5984f369..f64c27c296 100644
--- a/plc4go/internal/bacnetip/UDPCommunicationsModule.go
+++ b/plc4go/internal/bacnetip/UDPCommunicationsModule.go
@@ -59,11 +59,12 @@ func NewUDPActor(director *UDPDirector, peer string) *UDPActor {
return a
}
-func (a *UDPActor) idleTimeout() {
+func (a *UDPActor) idleTimeout() error {
log.Debug().Msg("idleTimeout")
// tell the director this is gone
a.director.DelActor(a)
+ return nil
}
func (a *UDPActor) Indication(pdu _PDU) error {