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 2022/11/17 16:46:08 UTC
[plc4x] branch develop updated: refactor(plc4go/bacnet): ported PDU object and encapsulate source and destination
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 789403bcb4 refactor(plc4go/bacnet): ported PDU object and encapsulate source and destination
789403bcb4 is described below
commit 789403bcb4ec45ac4a7f47c088440b11c3b838bd
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Thu Nov 17 17:45:49 2022 +0100
refactor(plc4go/bacnet): ported PDU object and encapsulate source and destination
---
plc4go/internal/bacnetip/ApplicationLayer.go | 545 +++++++++++----------
plc4go/internal/bacnetip/ApplicationModule.go | 52 +-
.../bacnetip/BACnetVirtualLinkLayerService.go | 160 ++++--
plc4go/internal/bacnetip/CommunicationsModule.go | 91 +++-
plc4go/internal/bacnetip/Driver.go | 6 +-
plc4go/internal/bacnetip/IOCBModule.go | 25 +-
plc4go/internal/bacnetip/MessageCodec.go | 12 +-
plc4go/internal/bacnetip/NetworkService.go | 82 +++-
plc4go/internal/bacnetip/PDU.go | 482 ++++++++++++++++++
.../internal/bacnetip/UDPCommunicationsModule.go | 7 +-
plc4go/internal/bacnetip/local/Device.go | 5 +
.../drivers/tests/manual_bacnet_driver_test.go | 11 +
12 files changed, 1077 insertions(+), 401 deletions(-)
diff --git a/plc4go/internal/bacnetip/ApplicationLayer.go b/plc4go/internal/bacnetip/ApplicationLayer.go
index ee33b885e8..d45f529f1d 100644
--- a/plc4go/internal/bacnetip/ApplicationLayer.go
+++ b/plc4go/internal/bacnetip/ApplicationLayer.go
@@ -20,10 +20,8 @@
package bacnetip
import (
- "bytes"
"github.com/apache/plc4x/plc4go/internal/bacnetip/local"
readWriteModel "github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model"
- "github.com/apache/plc4x/plc4go/spi"
"github.com/apache/plc4x/plc4go/spi/utils"
"github.com/pkg/errors"
"github.com/rs/zerolog/log"
@@ -81,7 +79,9 @@ type SSMSAPRequirements interface {
GetLocalDevice() *local.LocalDeviceObject
GetProposedWindowSize() uint8
GetClientTransactions() []*ClientSSM
+ RemoveClientTransaction(*ClientSSM)
GetServerTransactions() []*ServerSSM
+ RemoveServerTransaction(*ServerSSM)
GetApplicationTimeout() uint
}
@@ -91,7 +91,7 @@ type SSM struct {
ssmSAP SSMSAPRequirements
- pduAddress []byte
+ pduAddress Address
deviceInfo *DeviceInfo
invokeId uint8
@@ -116,10 +116,10 @@ type SSM struct {
maxApduLengthAccepted *readWriteModel.MaxApduLengthAccepted
}
-func NewSSM(sap SSMSAPRequirements, pduAddress []byte) (SSM, error) {
- log.Debug().Interface("sap", sap).Bytes("pdu_address", pduAddress).Msg("init")
+func NewSSM(sap SSMSAPRequirements, 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})
+ deviceInfoTemp, ok := sap.GetDeviceInfoCache().GetDeviceInfo(DeviceInfoCacheKey{PduSource: &pduAddress})
if ok {
deviceInfo = &deviceInfoTemp
}
@@ -221,7 +221,7 @@ func (s *SSM) setSegmentationContext(apdu readWriteModel.APDU) error {
// getSegment This function returns an APDU coorisponding to a particular segment of a confirmed request or complex ack.
// The segmentAPDU is the context
-func (s *SSM) getSegment(index uint8) (segmentAPDU readWriteModel.APDU, moreFollows bool, err error) {
+func (s *SSM) getSegment(index uint8) (segmentAPDU _PDU, moreFollows bool, err error) {
log.Debug().Msgf("Get segment %d", index)
if s.segmentAPDU == nil {
return nil, false, errors.New("No segment apdu set")
@@ -233,7 +233,7 @@ func (s *SSM) getSegment(index uint8) (segmentAPDU readWriteModel.APDU, moreFoll
// TODO: the original code does here something funky but it seems it is best to just return the original apdu
if s.segmentCount == 1 {
- return s.segmentAPDU.originalApdu, false, nil
+ return NewPDU(s.segmentAPDU.originalApdu, WithPDUDestination(s.pduAddress)), false, nil
}
moreFollows = index < s.segmentCount-1
@@ -251,7 +251,7 @@ func (s *SSM) getSegment(index uint8) (segmentAPDU readWriteModel.APDU, moreFoll
log.Debug().Msg("confirmed request context")
segmentedResponseAccepted := s.segmentationSupported == readWriteModel.BACnetSegmentation_SEGMENTED_RECEIVE || s.segmentationSupported == readWriteModel.BACnetSegmentation_SEGMENTED_BOTH
log.Debug().Msgf("segmentedResponseAccepted %t", segmentedResponseAccepted)
- segmentAPDU = readWriteModel.NewAPDUConfirmedRequest(
+ segmentAPDU = NewPDU(readWriteModel.NewAPDUConfirmedRequest(
true,
moreFollows,
segmentedResponseAccepted,
@@ -264,10 +264,10 @@ func (s *SSM) getSegment(index uint8) (segmentAPDU readWriteModel.APDU, moreFoll
serviceChoice,
segmentBytes,
0,
- )
+ ), WithPDUDestination(s.pduAddress))
} else {
log.Debug().Msg("complex ack context")
- segmentAPDU = readWriteModel.NewAPDUComplexAck(
+ segmentAPDU = NewPDU(readWriteModel.NewAPDUComplexAck(
true,
moreFollows,
s.segmentAPDU.originalInvokeId,
@@ -277,7 +277,7 @@ func (s *SSM) getSegment(index uint8) (segmentAPDU readWriteModel.APDU, moreFoll
serviceChoice,
segmentBytes,
0,
- )
+ ), WithPDUDestination(s.pduAddress))
}
return segmentAPDU, moreFollows, nil
}
@@ -285,7 +285,7 @@ func (s *SSM) getSegment(index uint8) (segmentAPDU readWriteModel.APDU, moreFoll
// TODO: check that function. looks a bit wonky to just append the payloads like that
// appendSegment This function appends the apdu content to the end of the current APDU being built. The segmentAPDU is
// the context
-func (s *SSM) appendSegment(apdu spi.Message) error {
+func (s *SSM) appendSegment(apdu _PDU) error {
log.Debug().Msgf("appendSegment\n%s", apdu)
switch apdu := apdu.(type) {
case readWriteModel.APDUConfirmedRequestExactly:
@@ -324,7 +324,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(apdu); err != nil {
+ if err := s.ssmSAP.Request(NewPDU(apdu, WithPDUDestination(s.pduAddress))); err != nil {
log.Debug().Err(err).Msg("error sending request")
}
if moreFollows {
@@ -338,8 +338,8 @@ type ClientSSM struct {
SSM
}
-func NewClientSSM(sap SSMSAPRequirements, pduAddress []byte) (*ClientSSM, error) {
- log.Debug().Interface("sap", sap).Bytes("pduAddress", pduAddress).Msg("init")
+func NewClientSSM(sap SSMSAPRequirements, pduAddress Address) (*ClientSSM, error) {
+ log.Debug().Interface("sap", sap).Interface("pduAddress", pduAddress).Msg("init")
ssm, err := NewSSM(sap, pduAddress)
if err != nil {
return nil, err
@@ -355,17 +355,17 @@ func NewClientSSM(sap SSMSAPRequirements, pduAddress []byte) (*ClientSSM, error)
}
// setState This function is called when the client wants to change state
-func (s *ClientSSM) setState(newState SSMState, timer *uint) error {
- log.Debug().Msgf("setState %s timer=%d", newState, timer)
+func (c *ClientSSM) setState(newState SSMState, timer *uint) error {
+ log.Debug().Msgf("setState %c timer=%d", newState, timer)
// do the regular state change
- if err := s.SSM.setState(newState, timer); err != nil {
+ if err := c.SSM.setState(newState, timer); err != nil {
return errors.Wrap(err, "error during SSM state transition")
}
- if s.state == SSMState_COMPLETED || s.state == SSMState_ABORTED {
+ if c.state == SSMState_COMPLETED || c.state == SSMState_ABORTED {
log.Debug().Msg("remove from active transaction")
- s.ssmSAP.GetClientTransactions() // TODO remove "this" transaction from the list
- if s.deviceInfo == nil {
+ c.ssmSAP.RemoveClientTransaction(c)
+ if c.deviceInfo == nil {
// TODO: release device entry
log.Debug().Msg("release device entry")
}
@@ -374,17 +374,21 @@ func (s *ClientSSM) setState(newState SSMState, timer *uint) error {
}
// Request This function is called by client transaction functions when it wants to send a message to the device
-func (s *ClientSSM) Request(apdu spi.Message) error {
- log.Debug().Msgf("request\n%s", apdu)
- // TODO: ensure apdu has destination, otherwise
- // TODO: we would need a BVLC to send something or not... maybe the todo above is nonsense, as we are in a connection context
- return s.ssmSAP.Request(apdu)
+func (c *ClientSSM) Request(apdu _PDU) error {
+ log.Debug().Msgf("request\n%c", apdu)
+
+ // make sure it has a good source and destination
+ nullAddress, _ := NewAddress()
+ apdu = NewPDUFromPDU(apdu, WithPDUSource(*nullAddress), WithPDUDestination(c.pduAddress))
+
+ // send it via the device
+ return c.ssmSAP.Request(apdu)
}
// Indication This function is called after the device has bound a new transaction and wants to start the process
// rolling
-func (s *ClientSSM) Indication(apdu spi.Message) error { // TODO: maybe use another name for that
- log.Debug().Msgf("indication\n%s", apdu)
+func (c *ClientSSM) Indication(apdu _PDU) error {
+ log.Debug().Msgf("indication\n%c", apdu)
// make sure we're getting confirmed requests
var apduConfirmedRequest readWriteModel.APDUConfirmedRequest
if apdu, ok := apdu.(readWriteModel.APDUConfirmedRequestExactly); !ok {
@@ -392,250 +396,251 @@ func (s *ClientSSM) Indication(apdu spi.Message) error { // TODO: maybe use anot
}
// save the request and set the segmentation context
- if err := s.setSegmentationContext(apduConfirmedRequest); err != nil {
+ if err := c.setSegmentationContext(apduConfirmedRequest); err != nil {
return errors.Wrap(err, "error setting context")
}
// if the max apdu length of the server isn't known, assume that it is the same size as our own and will be the segment
// size
- if s.deviceInfo == nil || s.deviceInfo.MaximumApduLengthAccepted != nil {
- s.segmentSize = uint(s.maxApduLengthAccepted.NumberOfOctets())
- } else if s.deviceInfo.MaximumNpduLength == nil {
+ if c.deviceInfo == nil || c.deviceInfo.MaximumApduLengthAccepted != nil {
+ c.segmentSize = uint(c.maxApduLengthAccepted.NumberOfOctets())
+ } else if c.deviceInfo.MaximumNpduLength == nil {
// if the max npdu length of the server isn't known, assume that it is the same as the max apdu length accepted
- s.segmentSize = uint(s.maxApduLengthAccepted.NumberOfOctets())
+ c.segmentSize = uint(c.maxApduLengthAccepted.NumberOfOctets())
} else {
- s.segmentSize = utils.Min(*s.deviceInfo.MaximumNpduLength, uint(s.maxApduLengthAccepted.NumberOfOctets()))
+ c.segmentSize = utils.Min(*c.deviceInfo.MaximumNpduLength, uint(c.maxApduLengthAccepted.NumberOfOctets()))
}
- log.Debug().Msgf("segment size %d", s.segmentSize)
+ log.Debug().Msgf("segment size %d", c.segmentSize)
- s.invokeId = apduConfirmedRequest.GetInvokeId()
- log.Debug().Msgf("invoke ID: %d", s.invokeId)
+ c.invokeId = apduConfirmedRequest.GetInvokeId()
+ log.Debug().Msgf("invoke ID: %d", c.invokeId)
var segmentCount, more int
- segmentCount, more = len(s.segmentAPDU.serviceBytes)/int(s.segmentSize), len(s.segmentAPDU.serviceBytes)%int(s.segmentSize)
- s.segmentCount = uint8(segmentCount)
+ segmentCount, more = len(c.segmentAPDU.serviceBytes)/int(c.segmentSize), len(c.segmentAPDU.serviceBytes)%int(c.segmentSize)
+ c.segmentCount = uint8(segmentCount)
if more > 0 {
- s.segmentCount += 1
+ c.segmentCount += 1
}
log.Debug().Msgf("segment count %d", segmentCount)
- if s.segmentCount > 1 {
- if s.segmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_TRANSMIT && s.segmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_BOTH {
+ if c.segmentCount > 1 {
+ if c.segmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_TRANSMIT && c.segmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_BOTH {
log.Debug().Msg("local device can't send segmented requests")
- abort, err := s.abort(readWriteModel.BACnetAbortReason_SEGMENTATION_NOT_SUPPORTED)
+ abort, err := c.abort(readWriteModel.BACnetAbortReason_SEGMENTATION_NOT_SUPPORTED)
if err != nil {
return errors.Wrap(err, "Error creating abort")
}
- return s.Response(abort)
+ return c.Response(abort)
}
- if s.deviceInfo == nil {
+ if c.deviceInfo == nil {
log.Debug().Msg("no server info for segmentation support")
- } else if *s.deviceInfo.SegmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_TRANSMIT && *s.deviceInfo.SegmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_BOTH {
+ } else if *c.deviceInfo.SegmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_TRANSMIT && *c.deviceInfo.SegmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_BOTH {
log.Debug().Msg("server can't receive segmented requests")
- abort, err := s.abort(readWriteModel.BACnetAbortReason_SEGMENTATION_NOT_SUPPORTED)
+ abort, err := c.abort(readWriteModel.BACnetAbortReason_SEGMENTATION_NOT_SUPPORTED)
if err != nil {
return errors.Wrap(err, "Error creating abort")
}
- return s.Response(abort)
+ return c.Response(abort)
}
// make sure we don't exceed the number of segments in our request that the server said it was willing to accept
- if s.deviceInfo == nil {
+ if c.deviceInfo == nil {
log.Debug().Msg("no server info for maximum number of segments")
- } else if s.deviceInfo.MaxSegmentsAccepted == nil {
+ } else if c.deviceInfo.MaxSegmentsAccepted == nil {
log.Debug().Msgf("server doesn't say maximum number of segments")
- } else if s.segmentCount > s.deviceInfo.MaxSegmentsAccepted.MaxSegments() {
+ } else if c.segmentCount > c.deviceInfo.MaxSegmentsAccepted.MaxSegments() {
log.Debug().Msg("server can't receive enough segments")
- abort, err := s.abort(readWriteModel.BACnetAbortReason_APDU_TOO_LONG)
+ abort, err := c.abort(readWriteModel.BACnetAbortReason_APDU_TOO_LONG)
if err != nil {
return errors.Wrap(err, "Error creating abort")
}
- return s.Response(abort)
+ return c.Response(abort)
}
}
// send out the first segment (or the whole thing)
- if s.segmentCount == 1 {
+ if c.segmentCount == 1 {
// unsegmented
- s.sentAllSegments = true
- s.retryCount = 0
- if err := s.setState(SSMState_AWAIT_CONFIRMATION, &s.apduTimeout); err != nil {
+ c.sentAllSegments = true
+ c.retryCount = 0
+ if err := c.setState(SSMState_AWAIT_CONFIRMATION, &c.apduTimeout); err != nil {
return errors.Wrap(err, "error switching state")
}
} else {
// segmented
- s.sentAllSegments = false
- s.retryCount = 0
- s.segmentRetryCount = 0
- s.initialSequenceNumber = 0
- s.actualWindowSize = nil
- if err := s.setState(SSMState_SEGMENTED_REQUEST, &s.segmentTimeout); err != nil {
+ c.sentAllSegments = false
+ c.retryCount = 0
+ c.segmentRetryCount = 0
+ c.initialSequenceNumber = 0
+ c.actualWindowSize = nil
+ if err := c.setState(SSMState_SEGMENTED_REQUEST, &c.segmentTimeout); err != nil {
return errors.Wrap(err, "error switching state")
}
}
// deliver to the device
- segment, _, err := s.getSegment(0)
+ segment, _, err := c.getSegment(0)
if err != nil {
return errors.Wrap(err, "error getting segment")
}
- return s.Request(segment)
+ return c.Request(segment)
}
// Response This function is called by client transaction functions when they want to send a message to the application.
-func (s *ClientSSM) Response(apdu spi.Message) error {
- log.Debug().Msgf("response\n%s", apdu)
+func (c *ClientSSM) Response(apdu _PDU) error {
+ log.Debug().Msgf("response\n%c", apdu)
+
// make sure it has a good source and destination
- // TODO: check if source == s.pduAddress
- // TODO: check if
+ nullAddress, _ := NewAddress()
+ apdu = NewPDUFromPDU(apdu, WithPDUSource(c.pduAddress), WithPDUDestination(*nullAddress))
// send it to the application
- return s.ssmSAP.SapResponse(apdu)
+ return c.ssmSAP.SapResponse(apdu)
}
// Confirmation This function is called by the device for all upstream messages related to the transaction.
-func (s *ClientSSM) Confirmation(apdu spi.Message) error {
- log.Debug().Msgf("confirmation\n%s", apdu)
+func (c *ClientSSM) Confirmation(apdu _PDU) error {
+ log.Debug().Msgf("confirmation\n%c", apdu)
- switch s.state {
+ switch c.state {
case SSMState_SEGMENTED_REQUEST:
- return s.segmentedRequest(apdu)
+ return c.segmentedRequest(apdu)
case SSMState_AWAIT_CONFIRMATION:
- return s.awaitConfirmation(apdu)
+ return c.awaitConfirmation(apdu)
case SSMState_SEGMENTED_CONFIRMATION:
- return s.segmentedConfirmation(apdu)
+ return c.segmentedConfirmation(apdu)
default:
- return errors.Errorf("Invalid state %s", s.state)
+ return errors.Errorf("Invalid state %c", c.state)
}
}
// processTask This function is called when something has taken too long
-func (s *ClientSSM) processTask() error {
+func (c *ClientSSM) processTask() error {
log.Debug().Msg("processTask")
- switch s.state {
+ switch c.state {
case SSMState_SEGMENTED_REQUEST:
- return s.segmentedRequestTimeout()
+ return c.segmentedRequestTimeout()
case SSMState_AWAIT_CONFIRMATION:
- return s.awaitConfirmationTimeout()
+ return c.awaitConfirmationTimeout()
case SSMState_SEGMENTED_CONFIRMATION:
- return s.segmentedConfirmationTimeout()
+ return c.segmentedConfirmationTimeout()
case SSMState_COMPLETED, SSMState_ABORTED:
return nil
default:
- return errors.Errorf("Invalid state %s", s.state)
+ return errors.Errorf("Invalid state %c", c.state)
}
}
// abort This function is called when the transaction should be aborted
-func (s *ClientSSM) abort(reason readWriteModel.BACnetAbortReason) (readWriteModel.APDU, error) {
- log.Debug().Msgf("abort\n%s", reason)
+func (c *ClientSSM) abort(reason readWriteModel.BACnetAbortReason) (_PDU, error) {
+ log.Debug().Msgf("abort\n%c", reason)
// change the state to aborted
- if err := s.setState(SSMState_ABORTED, nil); err != nil {
+ if err := c.setState(SSMState_ABORTED, nil); err != nil {
return nil, errors.Wrap(err, "Error setting state to aborted")
}
// build an abort PDU to return
- abortApdu := readWriteModel.NewAPDUAbort(false, s.invokeId, readWriteModel.NewBACnetAbortReasonTagged(reason, uint32(reason), 0), 0)
+ abortApdu := readWriteModel.NewAPDUAbort(false, c.invokeId, readWriteModel.NewBACnetAbortReasonTagged(reason, uint32(reason), 0), 0)
// return it
- return abortApdu, nil
+ return NewPDU(abortApdu), nil
}
// segmentedRequest This function is called when the client is sending a segmented request and receives an apdu
-func (s *ClientSSM) segmentedRequest(apdu spi.Message) error {
- log.Debug().Msgf("segmentedRequest\n%s", apdu)
+func (c *ClientSSM) segmentedRequest(apdu _PDU) error {
+ log.Debug().Msgf("segmentedRequest\n%c", apdu)
- switch apdu := apdu.(type) {
+ switch _apdu := apdu.(type) {
// server is ready for the next segment
case readWriteModel.APDUSegmentAckExactly:
log.Debug().Msg("segment ack")
- getActualWindowSize := apdu.GetActualWindowSize()
- s.actualWindowSize = &getActualWindowSize
+ getActualWindowSize := _apdu.GetActualWindowSize()
+ c.actualWindowSize = &getActualWindowSize
// duplicate ack received?
- if !s.inWindow(apdu.GetSequenceNumber(), s.initialSequenceNumber) {
+ if !c.inWindow(_apdu.GetSequenceNumber(), c.initialSequenceNumber) {
log.Debug().Msg("not in window")
- s.restartTimer(s.segmentTimeout)
- } else if s.sentAllSegments {
+ c.restartTimer(c.segmentTimeout)
+ } else if c.sentAllSegments {
log.Debug().Msg("all done sending request")
- if err := s.setState(SSMState_AWAIT_CONFIRMATION, &s.apduTimeout); err != nil {
+ if err := c.setState(SSMState_AWAIT_CONFIRMATION, &c.apduTimeout); err != nil {
return errors.Wrap(err, "error switching state")
}
} else {
log.Debug().Msg("More segments to send")
- s.initialSequenceNumber = apdu.GetSequenceNumber() + 1
- s.retryCount = 0
- if err := s.fillWindow(s.initialSequenceNumber); err != nil {
+ c.initialSequenceNumber = _apdu.GetSequenceNumber() + 1
+ c.retryCount = 0
+ if err := c.fillWindow(c.initialSequenceNumber); err != nil {
return errors.Wrap(err, "error filling window")
}
- s.restartTimer(s.segmentTimeout)
+ c.restartTimer(c.segmentTimeout)
}
// simple ack
case readWriteModel.APDUSimpleAckExactly:
log.Debug().Msg("simple ack")
- if !s.sentAllSegments {
- abort, err := s.abort(readWriteModel.BACnetAbortReason_INVALID_APDU_IN_THIS_STATE)
+ if !c.sentAllSegments {
+ abort, err := c.abort(readWriteModel.BACnetAbortReason_INVALID_APDU_IN_THIS_STATE)
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- if err := s.Request(abort); err != nil { // send it ot the device
+ if err := c.Request(abort); err != nil { // send it ot the device
log.Debug().Err(err).Msg("error sending request")
}
- if err := s.Response(abort); err != nil { // send it ot the application
+ if err := c.Response(abort); err != nil { // send it ot the application
log.Debug().Err(err).Msg("error sending response")
}
} else {
- if err := s.setState(SSMState_COMPLETED, nil); err != nil {
+ if err := c.setState(SSMState_COMPLETED, nil); err != nil {
return errors.Wrap(err, "error switching state")
}
}
// complex ack
case readWriteModel.APDUComplexAckExactly:
log.Debug().Msg("complex ack")
- if !s.sentAllSegments {
- abort, err := s.abort(readWriteModel.BACnetAbortReason_INVALID_APDU_IN_THIS_STATE)
+ if !c.sentAllSegments {
+ abort, err := c.abort(readWriteModel.BACnetAbortReason_INVALID_APDU_IN_THIS_STATE)
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- if err := s.Request(abort); err != nil { // send it ot the device
+ if err := c.Request(abort); err != nil { // send it ot the device
log.Debug().Err(err).Msg("error sending request")
}
- if err := s.Response(abort); err != nil { // send it ot the application
+ if err := c.Response(abort); err != nil { // send it ot the application
log.Debug().Err(err).Msg("error sending response")
}
- } else if !apdu.GetSegmentedMessage() {
+ } else if !_apdu.GetSegmentedMessage() {
// ack is not segmented
- if err := s.setState(SSMState_COMPLETED, nil); err != nil {
+ if err := c.setState(SSMState_COMPLETED, nil); err != nil {
return errors.Wrap(err, "error switching state")
}
- if err := s.Response(apdu); err != nil {
+ if err := c.Response(apdu); err != nil {
log.Debug().Err(err).Msg("error sending response")
}
} else {
// set the segmented response context
- if err := s.setSegmentationContext(apdu); err != nil {
+ if err := c.setSegmentationContext(_apdu); err != nil {
return errors.Wrap(err, "error setting context")
}
// minimum of what the server is proposing and this client proposes
- minWindowSize := utils.Min(*apdu.GetProposedWindowSize(), s.ssmSAP.GetProposedWindowSize())
- s.actualWindowSize = &minWindowSize
- s.lastSequenceNumber = 0
- s.initialSequenceNumber = 0
- if err := s.setState(SSMState_SEGMENTED_CONFIRMATION, &s.segmentTimeout); err != nil {
+ minWindowSize := utils.Min(*_apdu.GetProposedWindowSize(), c.ssmSAP.GetProposedWindowSize())
+ c.actualWindowSize = &minWindowSize
+ c.lastSequenceNumber = 0
+ c.initialSequenceNumber = 0
+ if err := c.setState(SSMState_SEGMENTED_CONFIRMATION, &c.segmentTimeout); err != nil {
return errors.Wrap(err, "error switching state")
}
}
case readWriteModel.APDUErrorExactly:
log.Debug().Msg("error/reject/abort")
- if err := s.setState(SSMState_COMPLETED, nil); err != nil {
+ if err := c.setState(SSMState_COMPLETED, nil); err != nil {
return errors.Wrap(err, "error switching state")
}
- if err := s.Response(apdu); err != nil {
+ if err := c.Response(apdu); err != nil {
log.Debug().Err(err).Msg("error sending response")
}
default:
@@ -644,260 +649,260 @@ func (s *ClientSSM) segmentedRequest(apdu spi.Message) error {
return nil
}
-func (s *ClientSSM) segmentedRequestTimeout() error {
+func (c *ClientSSM) segmentedRequestTimeout() error {
log.Debug().Msg("segmentedRequestTimeout")
// Try again
- if s.segmentRetryCount < s.numberOfApduRetries {
+ if c.segmentRetryCount < c.numberOfApduRetries {
log.Debug().Msg("retry segmented request")
- s.segmentRetryCount++
- s.startTimer(s.segmentTimeout)
+ c.segmentRetryCount++
+ c.startTimer(c.segmentTimeout)
- if s.initialSequenceNumber == 0 {
- apdu, _, err := s.getSegment(0)
+ if c.initialSequenceNumber == 0 {
+ apdu, _, err := c.getSegment(0)
if err != nil {
return errors.Wrap(err, "error getting first segment")
}
- if err := s.Request(apdu); err != nil {
+ if err := c.Request(apdu); err != nil {
log.Debug().Err(err).Msg("error sending request")
}
} else {
- if err := s.fillWindow(s.initialSequenceNumber); err != nil {
+ if err := c.fillWindow(c.initialSequenceNumber); err != nil {
return errors.Wrap(err, "error filling window")
}
}
} else {
log.Debug().Msg("abort, no response from the device")
- abort, err := s.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
+ 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 {
return errors.Wrap(err, "error creating abort")
}
- if err := s.Response(abort); err != nil {
+ if err := c.Response(abort); err != nil {
log.Debug().Err(err).Msg("error sending response")
}
}
return nil
}
-func (s *ClientSSM) awaitConfirmation(apdu spi.Message) error {
- log.Debug().Msgf("awaitConfirmation\n%s", apdu)
+func (c *ClientSSM) awaitConfirmation(apdu _PDU) error {
+ log.Debug().Msgf("awaitConfirmation\n%c", apdu)
- switch apdu := apdu.(type) {
+ switch _apdu := apdu.(type) {
case readWriteModel.APDUAbortExactly:
log.Debug().Msg("Server aborted")
- if err := s.setState(SSMState_ABORTED, nil); err != nil {
+ if err := c.setState(SSMState_ABORTED, nil); err != nil {
return errors.Wrap(err, "error switching state")
}
- if err := s.Response(apdu); err != nil {
+ if err := c.Response(apdu); err != nil {
log.Debug().Err(err).Msg("error sending response")
}
case readWriteModel.APDUSimpleAckExactly, readWriteModel.APDUErrorExactly, readWriteModel.APDURejectExactly:
log.Debug().Msg("simple ack, error or reject")
- if err := s.setState(SSMState_COMPLETED, nil); err != nil {
+ if err := c.setState(SSMState_COMPLETED, nil); err != nil {
return errors.Wrap(err, "error switching state")
}
- if err := s.Response(apdu); err != nil {
+ if err := c.Response(apdu); err != nil {
log.Debug().Err(err).Msg("error sending response")
}
case readWriteModel.APDUComplexAckExactly:
log.Debug().Msg("complex ack")
- if !apdu.GetSegmentedMessage() {
+ if !_apdu.GetSegmentedMessage() {
log.Debug().Msg("unsegmented")
- if err := s.setState(SSMState_COMPLETED, nil); err != nil {
+ if err := c.setState(SSMState_COMPLETED, nil); err != nil {
return errors.Wrap(err, "error switching state")
}
- if err := s.Response(apdu); err != nil {
+ if err := c.Response(apdu); err != nil {
log.Debug().Err(err).Msg("error sending response")
}
- } else if s.segmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_RECEIVE && s.segmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_BOTH {
+ } else if c.segmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_RECEIVE && c.segmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_BOTH {
log.Debug().Msg("local device can't receive segmented messages")
- abort, err := s.abort(readWriteModel.BACnetAbortReason_SEGMENTATION_NOT_SUPPORTED)
+ abort, err := c.abort(readWriteModel.BACnetAbortReason_SEGMENTATION_NOT_SUPPORTED)
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- if err := s.Response(abort); err != nil {
+ if err := c.Response(abort); err != nil {
log.Debug().Err(err).Msg("error sending response")
}
- } else if *apdu.GetSequenceNumber() == 0 {
+ } else if *_apdu.GetSequenceNumber() == 0 {
log.Debug().Msg("segmented response")
// set the segmented response context
- if err := s.setSegmentationContext(apdu); err != nil {
+ if err := c.setSegmentationContext(_apdu); err != nil {
return errors.Wrap(err, "error set segmentation context")
}
- s.actualWindowSize = apdu.GetProposedWindowSize()
- s.lastSequenceNumber = 0
- s.initialSequenceNumber = 0
- if err := s.setState(SSMState_SEGMENTED_CONFIRMATION, nil); err != nil {
+ c.actualWindowSize = _apdu.GetProposedWindowSize()
+ c.lastSequenceNumber = 0
+ c.initialSequenceNumber = 0
+ if err := c.setState(SSMState_SEGMENTED_CONFIRMATION, nil); err != nil {
return errors.Wrap(err, "error switching state")
}
// send back a segment ack
- segmentAck := readWriteModel.NewAPDUSegmentAck(false, false, s.invokeId, s.initialSequenceNumber, *s.actualWindowSize, 0)
- if err := s.Request(segmentAck); err != nil {
+ segmentAck := readWriteModel.NewAPDUSegmentAck(false, false, c.invokeId, c.initialSequenceNumber, *c.actualWindowSize, 0)
+ if err := c.Request(NewPDU(segmentAck)); err != nil {
log.Debug().Err(err).Msg("error sending request")
}
} else {
log.Debug().Msg("Invalid apdu in this state")
- abort, err := s.abort(readWriteModel.BACnetAbortReason_INVALID_APDU_IN_THIS_STATE)
+ abort, err := c.abort(readWriteModel.BACnetAbortReason_INVALID_APDU_IN_THIS_STATE)
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- if err := s.Request(abort); err != nil { // send it ot the device
+ if err := c.Request(abort); err != nil { // send it ot the device
log.Debug().Err(err).Msg("error sending request")
}
- if err := s.Response(abort); err != nil { // send it ot the application
+ if err := c.Response(abort); err != nil { // send it ot the application
log.Debug().Err(err).Msg("error sending response")
}
}
case readWriteModel.APDUSegmentAckExactly:
log.Debug().Msg("segment ack(!?)")
- s.restartTimer(s.segmentTimeout)
+ c.restartTimer(c.segmentTimeout)
default:
return errors.Errorf("invalid apdu %T", apdu)
}
return nil
}
-func (s *ClientSSM) awaitConfirmationTimeout() error {
+func (c *ClientSSM) awaitConfirmationTimeout() error {
log.Debug().Msg("awaitConfirmationTimeout")
- if s.retryCount < s.numberOfApduRetries {
- log.Debug().Msgf("no response, try again (%d < %d)", s.retryCount, s.numberOfApduRetries)
- s.retryCount++
+ if c.retryCount < c.numberOfApduRetries {
+ log.Debug().Msgf("no response, try again (%d < %d)", c.retryCount, c.numberOfApduRetries)
+ c.retryCount++
// save the retry count, indication acts like the request is coming from the application so the retryCount gets
// re-initialized.
- saveCount := s.retryCount
- if err := s.Indication(s.segmentAPDU.originalApdu); err != nil { // TODO: check that it is really the intention to re-send the original apdu here
+ saveCount := c.retryCount
+ if err := c.Indication(NewPDU(c.segmentAPDU.originalApdu, WithPDUDestination(c.pduAddress))); err != nil { // TODO: check that it is really the intention to re-send the original apdu here
return err
}
- s.retryCount = saveCount
+ c.retryCount = saveCount
} else {
log.Debug().Msg("retry count exceeded")
- abort, err := s.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
+ 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 {
return errors.Wrap(err, "error creating abort")
}
- if err := s.Response(abort); err != nil {
+ if err := c.Response(abort); err != nil {
log.Debug().Err(err).Msg("error sending response")
}
}
return nil
}
-func (s *ClientSSM) segmentedConfirmation(apdu spi.Message) error {
- log.Debug().Msgf("segmentedConfirmation\n%s", apdu)
+func (c *ClientSSM) segmentedConfirmation(apdu _PDU) error {
+ log.Debug().Msgf("segmentedConfirmation\n%c", apdu)
// the only messages we should be getting are complex acks
apduComplexAck, ok := apdu.(readWriteModel.APDUComplexAckExactly)
if !ok {
log.Debug().Msg("complex ack required")
- abort, err := s.abort(readWriteModel.BACnetAbortReason_INVALID_APDU_IN_THIS_STATE)
+ abort, err := c.abort(readWriteModel.BACnetAbortReason_INVALID_APDU_IN_THIS_STATE)
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- if err := s.Request(abort); err != nil { // send it ot the device
+ if err := c.Request(abort); err != nil { // send it ot the device
log.Debug().Err(err).Msg("error sending request")
}
- if err := s.Response(abort); err != nil { // send it ot the application
+ if err := c.Response(abort); err != nil { // send it ot the application
log.Debug().Err(err).Msg("error sending response")
}
}
// it must be segmented
if !apduComplexAck.GetSegmentedMessage() {
- abort, err := s.abort(readWriteModel.BACnetAbortReason_INVALID_APDU_IN_THIS_STATE)
+ abort, err := c.abort(readWriteModel.BACnetAbortReason_INVALID_APDU_IN_THIS_STATE)
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- if err := s.Request(abort); err != nil { // send it ot the device
+ if err := c.Request(abort); err != nil { // send it ot the device
log.Debug().Err(err).Msg("error sending request")
}
- if err := s.Response(abort); err != nil { // send it ot the application
+ if err := c.Response(abort); err != nil { // send it ot the application
log.Debug().Err(err).Msg("error sending response")
}
}
// proper segment number
- if *apduComplexAck.GetSequenceNumber() != s.lastSequenceNumber+1 {
- log.Debug().Msgf("segment %d received out of order, should be %d", apduComplexAck.GetSequenceNumber(), s.lastSequenceNumber+1)
+ if *apduComplexAck.GetSequenceNumber() != c.lastSequenceNumber+1 {
+ log.Debug().Msgf("segment %d received out of order, should be %d", apduComplexAck.GetSequenceNumber(), c.lastSequenceNumber+1)
// segment received out of order
- s.restartTimer(s.segmentTimeout)
- segmentAck := readWriteModel.NewAPDUSegmentAck(true, false, s.invokeId, s.initialSequenceNumber, *s.actualWindowSize, 0)
- if err := s.Request(segmentAck); err != nil {
+ c.restartTimer(c.segmentTimeout)
+ segmentAck := readWriteModel.NewAPDUSegmentAck(true, false, c.invokeId, c.initialSequenceNumber, *c.actualWindowSize, 0)
+ if err := c.Request(NewPDU(segmentAck)); err != nil {
log.Debug().Err(err).Msg("error sending request")
}
return nil
}
// add the data
- if err := s.appendSegment(apdu); err != nil {
+ if err := c.appendSegment(apdu); err != nil {
return errors.Wrap(err, "error appending the segment")
}
// update the sequence number
- s.lastSequenceNumber = s.lastSequenceNumber + 1
+ c.lastSequenceNumber = c.lastSequenceNumber + 1
// last segment received
if !apduComplexAck.GetMoreFollows() {
log.Debug().Msg("No more follows")
// send final ack
- segmentAck := readWriteModel.NewAPDUSegmentAck(false, false, s.invokeId, s.lastSequenceNumber, *s.actualWindowSize, 0)
- if err := s.Request(segmentAck); err != nil {
+ segmentAck := readWriteModel.NewAPDUSegmentAck(false, false, c.invokeId, c.lastSequenceNumber, *c.actualWindowSize, 0)
+ if err := c.Request(NewPDU(segmentAck)); err != nil {
log.Debug().Err(err).Msg("error sending request")
}
- if err := s.setState(SSMState_COMPLETED, nil); err != nil {
+ if err := c.setState(SSMState_COMPLETED, nil); err != nil {
return errors.Wrap(err, "error switching state")
}
// TODO: this is nonsense... We need to parse the service and the apdu not sure where to get it from now...
// TODO: it should be the original apdu, we might just need to use that as base and forward it as non segmented
- parse, err := readWriteModel.APDUParse(s.segmentAPDU.serviceBytes, uint16(len(s.segmentAPDU.serviceBytes)))
+ parse, err := readWriteModel.APDUParse(c.segmentAPDU.serviceBytes, uint16(len(c.segmentAPDU.serviceBytes)))
if err != nil {
return errors.Wrap(err, "error parsing apdu")
}
- if err := s.Response(parse); err != nil {
+ if err := c.Response(NewPDU(parse)); err != nil {
log.Debug().Err(err).Msg("error sending response")
}
- } else if *apduComplexAck.GetSequenceNumber() == s.initialSequenceNumber+*s.actualWindowSize {
+ } else if *apduComplexAck.GetSequenceNumber() == c.initialSequenceNumber+*c.actualWindowSize {
log.Debug().Msg("last segment in the group")
- s.initialSequenceNumber = s.lastSequenceNumber
- s.restartTimer(s.segmentTimeout)
- segmentAck := readWriteModel.NewAPDUSegmentAck(false, false, s.invokeId, s.lastSequenceNumber, *s.actualWindowSize, 0)
- if err := s.Request(segmentAck); err != nil { // send it ot the device
+ c.initialSequenceNumber = c.lastSequenceNumber
+ c.restartTimer(c.segmentTimeout)
+ segmentAck := readWriteModel.NewAPDUSegmentAck(false, false, c.invokeId, c.lastSequenceNumber, *c.actualWindowSize, 0)
+ if err := c.Request(NewPDU(segmentAck)); err != nil { // send it ot the device
log.Debug().Err(err).Msg("error sending request")
}
} else {
log.Debug().Msg("Wait for more segments")
- s.restartTimer(s.segmentTimeout)
+ c.restartTimer(c.segmentTimeout)
}
return nil
}
-func (s *ClientSSM) segmentedConfirmationTimeout() error {
+func (c *ClientSSM) segmentedConfirmationTimeout() error {
log.Debug().Msg("segmentedConfirmationTimeout")
- abort, err := s.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
+ 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 {
return errors.Wrap(err, "error creating abort")
}
- return s.Response(abort)
+ return c.Response(abort)
}
type ServerSSM struct {
@@ -905,8 +910,8 @@ type ServerSSM struct {
segmentedResponseAccepted bool
}
-func NewServerSSM(sap SSMSAPRequirements, pduAddress []byte) (*ServerSSM, error) {
- log.Debug().Interface("sap", sap).Bytes("pduAddress", pduAddress).Msg("init")
+func NewServerSSM(sap SSMSAPRequirements, pduAddress Address) (*ServerSSM, error) {
+ log.Debug().Interface("sap", sap).Interface("pduAddress", pduAddress).Msg("init")
ssm, err := NewSSM(sap, pduAddress)
if err != nil {
return nil, err
@@ -932,7 +937,7 @@ func (s *ServerSSM) setState(newState SSMState, timer *uint) error {
if s.state == SSMState_COMPLETED || s.state == SSMState_ABORTED {
log.Debug().Msg("remove from active transaction")
- s.ssmSAP.GetServerTransactions() // TODO remove "this" transaction from the list
+ s.ssmSAP.RemoveServerTransaction(s)
if s.deviceInfo != nil {
// TODO: release device entry
log.Debug().Msg("release device entry")
@@ -942,7 +947,7 @@ func (s *ServerSSM) setState(newState SSMState, timer *uint) error {
}
// Request This function is called by transaction functions to send to the application
-func (s *ServerSSM) Request(apdu spi.Message) error {
+func (s *ServerSSM) Request(apdu _PDU) error {
log.Debug().Msgf("request\n%s", apdu)
// TODO: ensure apdu has destination, otherwise
// TODO: we would need a BVLC to send something or not... maybe the todo above is nonsense, as we are in a connection context
@@ -951,7 +956,7 @@ func (s *ServerSSM) Request(apdu spi.Message) error {
// Indication This function is called for each downstream packet related to
// the transaction
-func (s *ServerSSM) Indication(apdu spi.Message) error { // TODO: maybe use another name for that
+func (s *ServerSSM) Indication(apdu _PDU) error { // TODO: maybe use another name for that
log.Debug().Msgf("indication\n%s", apdu)
// make sure we're getting confirmed requests
@@ -970,7 +975,7 @@ func (s *ServerSSM) Indication(apdu spi.Message) error { // TODO: maybe use anot
}
// Response This function is called by client transaction functions when they want to send a message to the application.
-func (s *ServerSSM) Response(apdu spi.Message) error {
+func (s *ServerSSM) Response(apdu _PDU) error {
log.Debug().Msgf("response\n%s", apdu)
// make sure it has a good source and destination
// TODO: check if source == none
@@ -982,7 +987,7 @@ func (s *ServerSSM) Response(apdu spi.Message) error {
// Confirmation This function is called when the application has provided a response and needs it to be sent to the
// client.
-func (s *ServerSSM) Confirmation(apdu spi.Message) error {
+func (s *ServerSSM) Confirmation(apdu _PDU) error {
log.Debug().Msgf("confirmation\n%s", apdu)
// check to see we are in the correct state
@@ -990,7 +995,7 @@ func (s *ServerSSM) Confirmation(apdu spi.Message) error {
log.Debug().Msg("warning: no expecting a response")
}
- switch apdu := apdu.(type) {
+ switch _apdu := apdu.(type) {
// abort response
case readWriteModel.APDUAbortExactly:
log.Debug().Msg("abort")
@@ -1017,7 +1022,7 @@ func (s *ServerSSM) Confirmation(apdu spi.Message) error {
log.Debug().Msg("complex ack")
// save the response and set the segmentation context
- if err := s.setSegmentationContext(apdu); err != nil {
+ if err := s.setSegmentationContext(_apdu); err != nil {
return errors.Wrap(err, "error settings segmentation context")
}
@@ -1030,7 +1035,7 @@ func (s *ServerSSM) Confirmation(apdu spi.Message) error {
}
// compute the segment count
- if len(apdu.GetSegment()) == 0 {
+ if len(_apdu.GetSegment()) == 0 {
// always at least one segment
s.segmentCount = 1
} else {
@@ -1130,7 +1135,7 @@ func (s *ServerSSM) processTask() error {
}
// abort This function is called when the transaction should be aborted
-func (s *ServerSSM) abort(reason readWriteModel.BACnetAbortReason) (readWriteModel.APDU, error) {
+func (s *ServerSSM) abort(reason readWriteModel.BACnetAbortReason) (_PDU, error) {
log.Debug().Msgf("abort\n%s", reason)
// change the state to aborted
@@ -1141,10 +1146,10 @@ func (s *ServerSSM) abort(reason readWriteModel.BACnetAbortReason) (readWriteMod
// build an abort PDU to return
abortApdu := readWriteModel.NewAPDUAbort(true, s.invokeId, readWriteModel.NewBACnetAbortReasonTagged(reason, uint32(reason), 0), 0)
// return it
- return abortApdu, nil
+ return NewPDU(abortApdu), nil
}
-func (s *ServerSSM) idle(apdu spi.Message) error {
+func (s *ServerSSM) idle(apdu _PDU) error {
log.Debug().Msgf("idle %s", apdu)
// make sure we're getting confirmed requests
@@ -1237,10 +1242,10 @@ func (s *ServerSSM) idle(apdu spi.Message) error {
// send back a segment ack
segack := readWriteModel.NewAPDUSegmentAck(false, true, s.invokeId, s.initialSequenceNumber, *s.actualWindowSize, 0)
log.Debug().Msgf("segAck: %s", segack)
- return s.Response(segack)
+ return s.Response(NewPDU(segack))
}
-func (s *ServerSSM) segmentedRequest(apdu spi.Message) error {
+func (s *ServerSSM) segmentedRequest(apdu _PDU) error {
log.Debug().Msgf("segmentedRequest\n%s", apdu)
// some kind of problem
@@ -1291,7 +1296,7 @@ func (s *ServerSSM) segmentedRequest(apdu spi.Message) error {
// send back a segment ack
segack := readWriteModel.NewAPDUSegmentAck(true, true, s.invokeId, s.initialSequenceNumber, *s.actualWindowSize, 0)
- return s.Response(segack)
+ return s.Response(NewPDU(segack))
}
// add the data
@@ -1308,7 +1313,7 @@ func (s *ServerSSM) segmentedRequest(apdu spi.Message) error {
// send back the final segment ack
segack := readWriteModel.NewAPDUSegmentAck(false, true, s.invokeId, s.lastSequenceNumber, *s.actualWindowSize, 0)
- if err := s.Response(segack); err != nil {
+ if err := s.Response(NewPDU(segack)); err != nil {
log.Debug().Err(err).Msg("error sending response")
}
@@ -1324,7 +1329,7 @@ func (s *ServerSSM) segmentedRequest(apdu spi.Message) error {
if err != nil {
return errors.Wrap(err, "error parsing apdu")
}
- if err := s.Request(parse); err != nil {
+ if err := s.Request(NewPDU(parse)); err != nil {
log.Debug().Err(err).Msg("error sending request")
}
} else if *apduConfirmedRequest.GetSequenceNumber() == s.initialSequenceNumber+*s.actualWindowSize {
@@ -1335,7 +1340,7 @@ func (s *ServerSSM) segmentedRequest(apdu spi.Message) error {
// send back a segment ack
segack := readWriteModel.NewAPDUSegmentAck(false, true, s.invokeId, s.initialSequenceNumber, *s.actualWindowSize, 0)
- if err := s.Response(segack); err != nil {
+ if err := s.Response(NewPDU(segack)); err != nil {
log.Debug().Err(err).Msg("error sending response")
}
} else {
@@ -1356,7 +1361,7 @@ func (s *ServerSSM) segmentedRequestTimeout() error {
return nil
}
-func (s *ServerSSM) awaitResponse(apdu spi.Message) error {
+func (s *ServerSSM) awaitResponse(apdu _PDU) error {
log.Debug().Msgf("awaitResponse\n%s", apdu)
switch apdu.(type) {
@@ -1393,20 +1398,20 @@ func (s *ServerSSM) awaitResponseTimeout() error {
return nil
}
-func (s *ServerSSM) segmentedResponse(apdu spi.Message) error {
+func (s *ServerSSM) segmentedResponse(apdu _PDU) error {
log.Debug().Msgf("segmentedResponse\n%s", apdu)
// client is ready for the next segment
- switch apdu := apdu.(type) {
+ switch _apdu := apdu.(type) {
case readWriteModel.APDUSegmentAckExactly:
log.Debug().Msg("segment ack")
// actual window size is provided by client
- getActualWindowSize := apdu.GetActualWindowSize()
+ getActualWindowSize := _apdu.GetActualWindowSize()
s.actualWindowSize = &getActualWindowSize
// duplicate ack received?
- if !s.inWindow(apdu.GetSequenceNumber(), s.initialSequenceNumber) {
+ if !s.inWindow(_apdu.GetSequenceNumber(), s.initialSequenceNumber) {
log.Debug().Msg("not in window")
s.restartTimer(s.segmentTimeout)
} else if s.sentAllSegments {
@@ -1418,8 +1423,8 @@ func (s *ServerSSM) segmentedResponse(apdu spi.Message) error {
} else {
log.Debug().Msg("more segments to send")
- s.initialSequenceNumber = apdu.GetSequenceNumber() + 1
- actualWindowSize := apdu.GetActualWindowSize()
+ s.initialSequenceNumber = _apdu.GetSequenceNumber() + 1
+ actualWindowSize := _apdu.GetActualWindowSize()
s.actualWindowSize = &actualWindowSize
s.segmentRetryCount = 0
if err := s.fillWindow(s.initialSequenceNumber); err != nil {
@@ -1528,7 +1533,7 @@ func NewStateMachineAccessPoint(localDevice *local.LocalDeviceObject, deviceInve
}
// getNextInvokeId Called by clients to get an unused invoke ID
-func (s *StateMachineAccessPoint) getNextInvokeId(address []byte) (uint8, error) {
+func (s *StateMachineAccessPoint) getNextInvokeId(address Address) (uint8, error) {
log.Debug().Msg("getNextInvokeId")
initialID := s.nextInvokeId
@@ -1547,15 +1552,16 @@ func (s *StateMachineAccessPoint) getNextInvokeId(address []byte) (uint8, error)
// TODO: double check that the logic here is right
for _, tr := range s.clientTransactions {
- if invokeId == tr.invokeId && bytes.Equal(address, tr.pduAddress) {
+ // TODO: replace deep equal
+ if invokeId == tr.invokeId && address.Equals(tr.pduAddress) {
return invokeId, nil
}
}
}
}
-// ConfirmationFromSource Packets coming up the stack are APDU's
-func (s *StateMachineAccessPoint) ConfirmationFromSource(apdu readWriteModel.APDU, pduSource []byte) error { // TODO: note we need a special method here as we don't contain src in the apdu
+// Confirmation Packets coming up the stack are APDU's
+func (s *StateMachineAccessPoint) Confirmation(apdu _PDU) error { // TODO: note we need a special method here as we don't contain src in the apdu
log.Debug().Msgf("confirmation\n%s", apdu)
// check device communication control
@@ -1563,14 +1569,17 @@ func (s *StateMachineAccessPoint) ConfirmationFromSource(apdu readWriteModel.APD
case readWriteModel.BACnetConfirmedServiceRequestDeviceCommunicationControlEnableDisable_ENABLE:
log.Debug().Msg("communications enabled")
case readWriteModel.BACnetConfirmedServiceRequestDeviceCommunicationControlEnableDisable_DISABLE:
+ apduType := apdu.(interface {
+ GetApduType() readWriteModel.ApduType
+ }).GetApduType()
switch {
- case apdu.GetApduType() == readWriteModel.ApduType_CONFIRMED_REQUEST_PDU &&
+ case apduType == readWriteModel.ApduType_CONFIRMED_REQUEST_PDU &&
apdu.(readWriteModel.APDUConfirmedRequest).GetServiceRequest().GetServiceChoice() == readWriteModel.BACnetConfirmedServiceChoice_DEVICE_COMMUNICATION_CONTROL:
log.Debug().Msg("continue with DCC request")
- case apdu.GetApduType() == readWriteModel.ApduType_CONFIRMED_REQUEST_PDU &&
+ case apduType == readWriteModel.ApduType_CONFIRMED_REQUEST_PDU &&
apdu.(readWriteModel.APDUConfirmedRequest).GetServiceRequest().GetServiceChoice() == readWriteModel.BACnetConfirmedServiceChoice_REINITIALIZE_DEVICE:
log.Debug().Msg("continue with reinitialize device")
- case apdu.GetApduType() == readWriteModel.ApduType_UNCONFIRMED_REQUEST_PDU &&
+ case apduType == readWriteModel.ApduType_UNCONFIRMED_REQUEST_PDU &&
apdu.(readWriteModel.APDUUnconfirmedRequest).GetServiceRequest().GetServiceChoice() == readWriteModel.BACnetUnconfirmedServiceChoice_WHO_IS:
log.Debug().Msg("continue with Who-Is")
default:
@@ -1581,12 +1590,14 @@ func (s *StateMachineAccessPoint) ConfirmationFromSource(apdu readWriteModel.APD
log.Debug().Msg("initiation disabled")
}
- switch apdu := apdu.(type) {
+ var pduSource = apdu.GetPDUSource()
+
+ switch _apdu := apdu.(type) {
case readWriteModel.APDUConfirmedRequestExactly:
// Find duplicates of this request
var tr *ServerSSM
for _, serverTransactionElement := range s.serverTransactions {
- if apdu.GetInvokeId() == serverTransactionElement.invokeId && bytes.Equal(pduSource, serverTransactionElement.pduAddress) {
+ if _apdu.GetInvokeId() == serverTransactionElement.invokeId && pduSource.Equals(serverTransactionElement.pduAddress) {
tr = serverTransactionElement
break
}
@@ -1614,7 +1625,7 @@ func (s *StateMachineAccessPoint) ConfirmationFromSource(apdu readWriteModel.APD
// find the client transaction this is acking
var tr *ClientSSM
for _, tr := range s.clientTransactions {
- if apdu.(interface{ GetOriginalInvokeId() uint8 }).GetOriginalInvokeId() == tr.invokeId && bytes.Equal(pduSource, tr.pduAddress) {
+ if apdu.(interface{ GetOriginalInvokeId() uint8 }).GetOriginalInvokeId() == tr.invokeId && pduSource.Equals(tr.pduAddress) {
break
}
}
@@ -1629,10 +1640,10 @@ func (s *StateMachineAccessPoint) ConfirmationFromSource(apdu readWriteModel.APD
}
case readWriteModel.APDUAbortExactly:
// find the transaction being aborted
- if apdu.GetServer() {
+ if _apdu.GetServer() {
var tr *ClientSSM
for _, tr := range s.clientTransactions {
- if apdu.(interface{ GetOriginalInvokeId() uint8 }).GetOriginalInvokeId() == tr.invokeId && bytes.Equal(pduSource, tr.pduAddress) {
+ if apdu.(interface{ GetOriginalInvokeId() uint8 }).GetOriginalInvokeId() == tr.invokeId && pduSource.Equals(tr.pduAddress) {
break
}
}
@@ -1648,7 +1659,7 @@ func (s *StateMachineAccessPoint) ConfirmationFromSource(apdu readWriteModel.APD
} else {
var tr *ServerSSM
for _, serverTransactionElement := range s.serverTransactions {
- if apdu.GetOriginalInvokeId() == serverTransactionElement.invokeId && bytes.Equal(pduSource, serverTransactionElement.pduAddress) {
+ if _apdu.GetOriginalInvokeId() == serverTransactionElement.invokeId && pduSource.Equals(serverTransactionElement.pduAddress) {
tr = serverTransactionElement
break
}
@@ -1665,10 +1676,10 @@ func (s *StateMachineAccessPoint) ConfirmationFromSource(apdu readWriteModel.APD
}
case readWriteModel.APDUSegmentAckExactly:
// find the transaction being aborted
- if apdu.GetServer() {
+ if _apdu.GetServer() {
var tr *ClientSSM
for _, tr := range s.clientTransactions {
- if apdu.(interface{ GetOriginalInvokeId() uint8 }).GetOriginalInvokeId() == tr.invokeId && bytes.Equal(pduSource, tr.pduAddress) {
+ if apdu.(interface{ GetOriginalInvokeId() uint8 }).GetOriginalInvokeId() == tr.invokeId && pduSource.Equals(tr.pduAddress) {
break
}
}
@@ -1684,7 +1695,7 @@ func (s *StateMachineAccessPoint) ConfirmationFromSource(apdu readWriteModel.APD
} else {
var tr *ServerSSM
for _, serverTransactionElement := range s.serverTransactions {
- if apdu.GetOriginalInvokeId() == serverTransactionElement.invokeId && bytes.Equal(pduSource, serverTransactionElement.pduAddress) {
+ if _apdu.GetOriginalInvokeId() == serverTransactionElement.invokeId && pduSource.Equals(serverTransactionElement.pduAddress) {
tr = serverTransactionElement
break
}
@@ -1706,12 +1717,10 @@ func (s *StateMachineAccessPoint) ConfirmationFromSource(apdu readWriteModel.APD
}
// SapIndication This function is called when the application is requesting a new transaction as a client.
-func (s *StateMachineAccessPoint) SapIndication(apdu spi.Message) error {
+func (s *StateMachineAccessPoint) SapIndication(apdu _PDU) error {
log.Debug().Msgf("sapIndication\n%s", apdu)
- // TODO: extract from somewhere
- var pduDestination []byte
- panic("we need pduDestination")
+ pduDestination := apdu.GetPDUDestination()
// check device communication control
switch s.dccEnableDisable {
@@ -1731,7 +1740,7 @@ func (s *StateMachineAccessPoint) SapIndication(apdu spi.Message) error {
}
}
- switch apdu := apdu.(type) {
+ switch _apdu := apdu.(type) {
case readWriteModel.APDUUnconfirmedRequestExactly:
// deliver to the device
if err := s.Request(apdu); err != nil {
@@ -1742,7 +1751,7 @@ func (s *StateMachineAccessPoint) SapIndication(apdu spi.Message) error {
// TODO: here it is getting slightly different: usually we give the invoke id from the outside as it is build already. So maybe we need to adjust that (we never create it, we need to check for collisions but maybe we should change that so we move the creation down here)
// s.getNextInvokeId()...
for _, tr := range s.clientTransactions {
- if apdu.GetInvokeId() == tr.invokeId && bytes.Equal(pduDestination, tr.pduAddress) {
+ if _apdu.GetInvokeId() == tr.invokeId && pduDestination.Equals(tr.pduAddress) {
return errors.New("invoke ID in use")
}
}
@@ -1772,17 +1781,15 @@ func (s *StateMachineAccessPoint) SapIndication(apdu spi.Message) error {
// SapConfirmation This function is called when the application is responding to a request, the apdu may be a simple
// ack, complex ack, error, reject or abort
-func (s *StateMachineAccessPoint) SapConfirmation(apdu spi.Message) error {
+func (s *StateMachineAccessPoint) SapConfirmation(apdu _PDU) error {
log.Debug().Msgf("sapConfirmation\n%s", apdu)
- // TODO: extract from somewhere
- var pduDestination []byte
- panic("we need pduDestination")
+ pduDestination := apdu.GetPDUDestination()
switch apdu.(type) {
case readWriteModel.APDUSimpleAckExactly, readWriteModel.APDUComplexAckExactly, readWriteModel.APDUErrorExactly, readWriteModel.APDURejectExactly:
// find the client transaction this is acking
var tr *ServerSSM
for _, tr := range s.serverTransactions {
- if apdu.(interface{ GetOriginalInvokeId() uint8 }).GetOriginalInvokeId() == tr.invokeId && bytes.Equal(pduDestination, tr.pduAddress) {
+ if apdu.(interface{ GetOriginalInvokeId() uint8 }).GetOriginalInvokeId() == tr.invokeId && pduDestination.Equals(tr.pduAddress) {
break
}
}
@@ -1817,10 +1824,36 @@ func (s *StateMachineAccessPoint) GetClientTransactions() []*ClientSSM {
return s.clientTransactions
}
+func (s *StateMachineAccessPoint) RemoveClientTransaction(c *ClientSSM) {
+ indexFound := -1
+ for i, tr := range s.clientTransactions {
+ if tr == c {
+ indexFound = i
+ break
+ }
+ }
+ if indexFound >= 0 {
+ s.clientTransactions = append(s.clientTransactions[:indexFound], s.clientTransactions[indexFound+1:]...)
+ }
+}
+
func (s *StateMachineAccessPoint) GetServerTransactions() []*ServerSSM {
return s.serverTransactions
}
+func (s *StateMachineAccessPoint) RemoveServerTransaction(sssm *ServerSSM) {
+ indexFound := -1
+ for i, tr := range s.serverTransactions {
+ if tr == sssm {
+ indexFound = i
+ break
+ }
+ }
+ if indexFound >= 0 {
+ s.serverTransactions = append(s.serverTransactions[:indexFound], s.serverTransactions[indexFound+1:]...)
+ }
+}
+
func (s *StateMachineAccessPoint) GetApplicationTimeout() uint {
return s.applicationTimeout
}
@@ -1846,14 +1879,14 @@ func NewApplicationServiceAccessPoint(aseID *int, sapID *int) (*ApplicationServi
}
// TODO: big WIP
-func (a *ApplicationServiceAccessPoint) Indication(apdu spi.Message) error {
+func (a *ApplicationServiceAccessPoint) Indication(apdu _PDU) error {
log.Debug().Msgf("Indication\n%s", apdu)
- switch apdu := apdu.(type) {
+ switch _apdu := apdu.(type) {
case readWriteModel.APDUConfirmedRequestExactly:
//assume no errors found
var errorFound error
- if !readWriteModel.BACnetConfirmedServiceChoiceKnows(uint8(apdu.GetServiceRequest().GetServiceChoice())) {
+ if !readWriteModel.BACnetConfirmedServiceChoiceKnows(uint8(_apdu.GetServiceRequest().GetServiceChoice())) {
errorFound = errors.New("unrecognized service")
}
@@ -1870,12 +1903,12 @@ func (a *ApplicationServiceAccessPoint) Indication(apdu spi.Message) error {
log.Debug().Err(errorFound).Msg("got error")
// TODO: map it to a error... code temporary placeholder
- return a.Response(readWriteModel.NewAPDUReject(apdu.GetInvokeId(), nil, 0))
+ return a.Response(NewPDU(readWriteModel.NewAPDUReject(_apdu.GetInvokeId(), nil, 0)))
}
case readWriteModel.APDUUnconfirmedRequestExactly:
//assume no errors found
var errorFound error
- if !readWriteModel.BACnetUnconfirmedServiceChoiceKnows(uint8(apdu.GetServiceRequest().GetServiceChoice())) {
+ if !readWriteModel.BACnetUnconfirmedServiceChoiceKnows(uint8(_apdu.GetServiceRequest().GetServiceChoice())) {
errorFound = errors.New("unrecognized service")
}
@@ -1899,7 +1932,7 @@ func (a *ApplicationServiceAccessPoint) Indication(apdu spi.Message) error {
}
// TODO: big WIP
-func (a *ApplicationServiceAccessPoint) SapIndication(apdu spi.Message) error {
+func (a *ApplicationServiceAccessPoint) SapIndication(apdu _PDU) error {
log.Debug().Msgf("SapIndication\n%s", apdu)
// TODO: check if we need to check apdu here
@@ -1908,7 +1941,7 @@ func (a *ApplicationServiceAccessPoint) SapIndication(apdu spi.Message) error {
}
// TODO: big WIP
-func (a *ApplicationServiceAccessPoint) Confirmation(apdu spi.Message) error {
+func (a *ApplicationServiceAccessPoint) Confirmation(apdu _PDU) error {
log.Debug().Msgf("Confirmation\n%s", apdu)
// TODO: check if we need to check apdu here
@@ -1917,7 +1950,7 @@ func (a *ApplicationServiceAccessPoint) Confirmation(apdu spi.Message) error {
}
// TODO: big WIP
-func (a *ApplicationServiceAccessPoint) SapConfirmation(apdu spi.Message) error {
+func (a *ApplicationServiceAccessPoint) SapConfirmation(apdu _PDU) error {
log.Debug().Msgf("SapConfirmation\n%s", apdu)
// TODO: check if we need to check apdu here
diff --git a/plc4go/internal/bacnetip/ApplicationModule.go b/plc4go/internal/bacnetip/ApplicationModule.go
index 7d3bab5dfe..60dc145246 100644
--- a/plc4go/internal/bacnetip/ApplicationModule.go
+++ b/plc4go/internal/bacnetip/ApplicationModule.go
@@ -20,13 +20,11 @@
package bacnetip
import (
- "bytes"
"encoding/binary"
"fmt"
"github.com/apache/plc4x/plc4go/internal/bacnetip/local"
"github.com/apache/plc4x/plc4go/internal/bacnetip/service"
readWriteModel "github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model"
- "github.com/apache/plc4x/plc4go/spi"
"github.com/pkg/errors"
"github.com/rs/zerolog/log"
"hash/fnv"
@@ -35,7 +33,7 @@ import (
type DeviceInfo struct {
DeviceIdentifier readWriteModel.BACnetTagPayloadObjectIdentifier
- Address []byte
+ Address Address
MaximumApduLengthAccepted *readWriteModel.MaxApduLengthAccepted
SegmentationSupported *readWriteModel.BACnetSegmentation
@@ -47,7 +45,7 @@ type DeviceInfo struct {
_cacheKey DeviceInfoCacheKey
}
-func NewDeviceInfo(deviceIdentifier readWriteModel.BACnetTagPayloadObjectIdentifier, address []byte) *DeviceInfo {
+func NewDeviceInfo(deviceIdentifier readWriteModel.BACnetTagPayloadObjectIdentifier, address Address) *DeviceInfo {
return &DeviceInfo{
DeviceIdentifier: deviceIdentifier,
Address: address,
@@ -66,7 +64,7 @@ func NewDeviceInfo(deviceIdentifier readWriteModel.BACnetTagPayloadObjectIdentif
// DeviceInfoCacheKey caches by either Instance, PduSource of both
type DeviceInfoCacheKey struct {
Instance *uint32
- PduSource []byte
+ PduSource *Address
}
func (k DeviceInfoCacheKey) HashKey() uint32 {
@@ -74,12 +72,12 @@ func (k DeviceInfoCacheKey) HashKey() uint32 {
if k.Instance != nil {
_ = binary.Write(h, binary.BigEndian, *k.Instance)
}
- _, _ = h.Write(k.PduSource)
+ _ = binary.Write(h, binary.BigEndian, k.PduSource)
return h.Sum32()
}
func (k DeviceInfoCacheKey) String() string {
- return fmt.Sprintf("key: %d/%x", k.Instance, k.PduSource)
+ return fmt.Sprintf("key: %d/%v", k.Instance, k.PduSource)
}
type DeviceInfoCache struct {
@@ -93,7 +91,7 @@ func NewDeviceInfoCache() *DeviceInfoCache {
}
func (d *DeviceInfoCache) String() string {
- return fmt.Sprintf("%#q", d)
+ return fmt.Sprintf("DeviceInfoCache(%d)", len(d.cache))
}
// HasDeviceInfo Return true if cache has information about the device.
@@ -103,7 +101,7 @@ func (d *DeviceInfoCache) HasDeviceInfo(key DeviceInfoCacheKey) bool {
}
// IAmDeviceInfo Create a device information record based on the contents of an IAmRequest and put it in the cache.
-func (d *DeviceInfoCache) IAmDeviceInfo(iAm readWriteModel.BACnetUnconfirmedServiceRequestIAm, pduSource []byte) {
+func (d *DeviceInfoCache) IAmDeviceInfo(iAm readWriteModel.BACnetUnconfirmedServiceRequestIAm, pduSource Address) {
log.Debug().Msgf("IAmDeviceInfo\n%s", iAm)
deviceIdentifier := iAm.GetDeviceIdentifier()
@@ -115,7 +113,7 @@ func (d *DeviceInfoCache) IAmDeviceInfo(iAm readWriteModel.BACnetUnconfirmedServ
// maybe there is a record for this address
if !ok {
- deviceInfo, ok = d.cache[DeviceInfoCacheKey{nil, pduSource}.HashKey()]
+ deviceInfo, ok = d.cache[DeviceInfoCacheKey{nil, &pduSource}.HashKey()]
}
// make a new one using the class provided
@@ -163,8 +161,8 @@ func (d *DeviceInfoCache) UpdateDeviceInfo(deviceInfo DeviceInfo) {
delete(d.cache, cacheKey.HashKey())
d.cache[DeviceInfoCacheKey{Instance: &instanceNumber}.HashKey()] = deviceInfo
}
- if bytes.Compare(deviceInfo.Address, cacheKey.PduSource) != 0 {
- cacheKey.PduSource = deviceInfo.Address
+ if !deviceInfo.Address.Equals(cacheKey.PduSource) {
+ cacheKey.PduSource = &deviceInfo.Address
delete(d.cache, cacheKey.HashKey())
d.cache[DeviceInfoCacheKey{PduSource: cacheKey.PduSource}.HashKey()] = deviceInfo
}
@@ -173,7 +171,7 @@ func (d *DeviceInfoCache) UpdateDeviceInfo(deviceInfo DeviceInfo) {
instanceNumber := deviceInfo.DeviceIdentifier.GetInstanceNumber()
deviceInfo._cacheKey = DeviceInfoCacheKey{
Instance: &instanceNumber,
- PduSource: deviceInfo.Address,
+ PduSource: &deviceInfo.Address,
}
d.cache[deviceInfo._cacheKey.HashKey()] = deviceInfo
}
@@ -216,11 +214,11 @@ type Application struct {
localDevice *local.LocalDeviceObject
deviceInfoCache *DeviceInfoCache
controllers map[string]interface{}
- helpers map[string]func(pdu spi.Message) error
+ helpers map[string]func(pdu _PDU) error
}
-func NewApplication(localDevice *local.LocalDeviceObject, localAddress net.Addr, deviceInfoCache *DeviceInfoCache, aseID *int) (*Application, error) {
- log.Debug().Msgf("NewApplication %v %s deviceInfoCache=%s aseID=%d", localDevice, localAddress, deviceInfoCache, aseID)
+func NewApplication(localDevice *local.LocalDeviceObject, localAddress Address, deviceInfoCache *DeviceInfoCache, aseID *int) (*Application, error) {
+ log.Debug().Msgf("NewApplication localDevice=%v localAddress=%v deviceInfoCache=%s aseID=%d", localDevice, &localAddress, deviceInfoCache, aseID)
a := &Application{}
var err error
a.ApplicationServiceElement, err = NewApplicationServiceElement(aseID, a)
@@ -261,7 +259,7 @@ func NewApplication(localDevice *local.LocalDeviceObject, localAddress net.Addr,
return a, nil
}
-func (a *Application) Request(apdu spi.Message) error {
+func (a *Application) Request(apdu _PDU) error {
log.Debug().Msgf("Request\n%s", apdu)
// double-check the input is the right kind of APDU
@@ -273,7 +271,7 @@ func (a *Application) Request(apdu spi.Message) error {
return a.ApplicationServiceElement.Request(apdu)
}
-func (a *Application) Indication(apdu spi.Message) error {
+func (a *Application) Indication(apdu _PDU) error {
log.Debug().Msgf("Indication\n%s", apdu)
// get a helper function
@@ -292,7 +290,7 @@ func (a *Application) Indication(apdu spi.Message) error {
if err := helperFn(apdu); err != nil {
log.Debug().Err(err).Msgf("err result")
// TODO: do proper mapping
- a.Response(readWriteModel.NewAPDUError(0, readWriteModel.BACnetConfirmedServiceChoice_CREATE_OBJECT, nil, 0))
+ a.Response(NewPDU(readWriteModel.NewAPDUError(0, readWriteModel.BACnetConfirmedServiceChoice_CREATE_OBJECT, nil, 0)))
}
return nil
@@ -305,7 +303,7 @@ type ApplicationIOController struct {
queueByAddress map[string]SieveQueue
}
-func NewApplicationIOController(localDevice *local.LocalDeviceObject, localAddress net.Addr, deviceInfoCache *DeviceInfoCache, aseID *int) (*ApplicationIOController, error) {
+func NewApplicationIOController(localDevice *local.LocalDeviceObject, localAddress Address, deviceInfoCache *DeviceInfoCache, aseID *int) (*ApplicationIOController, error) {
a := &ApplicationIOController{
// queues for each address
queueByAddress: make(map[string]SieveQueue),
@@ -342,7 +340,7 @@ func (a *ApplicationIOController) ProcessIO(iocb _IOCB) error {
return queue.RequestIO(iocb)
}
-func (a *ApplicationIOController) _AppComplete(address net.Addr, apdu spi.Message) error {
+func (a *ApplicationIOController) _AppComplete(address net.Addr, apdu _PDU) error {
log.Debug().Msgf("_AppComplete %s\n%s", address, apdu)
// look up the queue
@@ -377,7 +375,7 @@ func (a *ApplicationIOController) _AppComplete(address net.Addr, apdu spi.Messag
return nil
}
-func (a *ApplicationIOController) _AppRequest(apdu spi.Message) {
+func (a *ApplicationIOController) _AppRequest(apdu _PDU) {
log.Debug().Msgf("_AppRequest\n%s", apdu)
if err := a.Request(apdu); err != nil {
@@ -398,7 +396,7 @@ func (a *ApplicationIOController) _AppRequest(apdu spi.Message) {
}
}
-func (a *ApplicationIOController) Request(apdu spi.Message) error {
+func (a *ApplicationIOController) Request(apdu _PDU) error {
log.Debug().Msgf("Request\n%s", apdu)
// if this is not unconfirmed request, tell the application to use the IOCB interface
@@ -410,7 +408,7 @@ func (a *ApplicationIOController) Request(apdu spi.Message) error {
return a.Application.Request(apdu)
}
-func (a *ApplicationIOController) Confirmation(apdu spi.Message) error {
+func (a *ApplicationIOController) Confirmation(apdu _PDU) error {
log.Debug().Msgf("Confirmation\n%s", apdu)
// this is an ack, error, reject or abort
@@ -423,7 +421,7 @@ type BIPSimpleApplication struct {
*ApplicationIOController
*service.WhoIsIAmServices
*service.ReadWritePropertyServices
- localAddress net.Addr
+ localAddress Address
asap *ApplicationServiceAccessPoint
smap *StateMachineAccessPoint
nsap *NetworkServiceAccessPoint
@@ -433,7 +431,7 @@ type BIPSimpleApplication struct {
mux *UDPMultiplexer
}
-func NewBIPSimpleApplication(localDevice *local.LocalDeviceObject, localAddress net.Addr, deviceInfoCache *DeviceInfoCache, aseID *int) (*BIPSimpleApplication, error) {
+func NewBIPSimpleApplication(localDevice *local.LocalDeviceObject, localAddress Address, deviceInfoCache *DeviceInfoCache, aseID *int) (*BIPSimpleApplication, error) {
b := &BIPSimpleApplication{}
var err error
b.ApplicationIOController, err = NewApplicationIOController(localDevice, localAddress, deviceInfoCache, aseID)
@@ -498,7 +496,7 @@ func NewBIPSimpleApplication(localDevice *local.LocalDeviceObject, localAddress
}
// bind the BIP stack to the network, no network number
- if err := b.nsap.bind(b.bip, nil, b.localAddress); err != nil {
+ if err := b.nsap.bind(b.bip, nil, &b.localAddress); err != nil {
return nil, err
}
diff --git a/plc4go/internal/bacnetip/BACnetVirtualLinkLayerService.go b/plc4go/internal/bacnetip/BACnetVirtualLinkLayerService.go
index ddd742e689..70104186ec 100644
--- a/plc4go/internal/bacnetip/BACnetVirtualLinkLayerService.go
+++ b/plc4go/internal/bacnetip/BACnetVirtualLinkLayerService.go
@@ -20,10 +20,8 @@
package bacnetip
import (
- "github.com/apache/plc4x/plc4go/spi"
"github.com/pkg/errors"
"github.com/rs/zerolog/log"
- "net"
)
type _MultiplexClient struct {
@@ -43,8 +41,8 @@ func _New_MultiplexClient(multiplexer *UDPMultiplexer) (*_MultiplexClient, error
return m, nil
}
-func (m *_MultiplexClient) Confirmation(pdu spi.Message) error {
- return m.multiplexer.Confirmation(pdu)
+func (m *_MultiplexClient) Confirmation(pdu _PDU) error {
+ return m.multiplexer.Confirmation(m, pdu)
}
type _MultiplexServer struct {
@@ -64,41 +62,66 @@ func _New_MultiplexServer(multiplexer *UDPMultiplexer) (*_MultiplexServer, error
return m, nil
}
-func (m *_MultiplexServer) Indication(pdu spi.Message) error {
- return m.multiplexer.Indication(pdu)
+func (m *_MultiplexServer) Indication(pdu _PDU) error {
+ return m.multiplexer.Indication(m, pdu)
}
type UDPMultiplexer struct {
- address *net.UDPAddr
- selfBroadcastAddress *net.UDPAddr
- direct *_MultiplexClient
- directPort *UDPDirector
- broadcast *_MultiplexClient
- broadcastPort *UDPDirector
- annexH *_MultiplexServer
- annexJ *_MultiplexServer
+ address Address
+ addrTuple *AddressTuple[string, uint16]
+ addrBroadcastTuple *AddressTuple[string, uint16]
+ direct *_MultiplexClient
+ directPort *UDPDirector
+ broadcast *_MultiplexClient
+ broadcastPort *UDPDirector
+ annexH *_MultiplexServer
+ annexJ *_MultiplexServer
}
-func NewUDPMultiplexer(address net.Addr, noBroadcast bool) (*UDPMultiplexer, error) {
+func NewUDPMultiplexer(address interface{}, noBroadcast bool) (*UDPMultiplexer, error) {
log.Debug().Msgf("NewUDPMultiplexer %v noBroadcast=%t", address, noBroadcast)
u := &UDPMultiplexer{}
// check for some options
specialBroadcast := false
if address == nil {
- u.address = &net.UDPAddr{IP: nil, Port: 46808}
- u.selfBroadcastAddress = &net.UDPAddr{IP: net.IPv4(255, 255, 255, 255), Port: 47808}
+ address, _ := NewAddress()
+ u.address = *address
+ u.addrTuple = &AddressTuple[string, uint16]{"", 47808}
+ u.addrBroadcastTuple = &AddressTuple[string, uint16]{"255.255.255.255", 47808}
} else {
- udpAddr, err := net.ResolveUDPAddr("udp", address.String())
- if err != nil {
- return nil, errors.Wrap(err, "error resolving upd")
+ // allow the address to be cast
+ if caddress, ok := address.(*Address); ok {
+ u.address = *caddress
+ } else if caddress, ok := address.(Address); ok {
+ u.address = caddress
+ } else {
+ newAddress, err := NewAddress(address)
+ if err != nil {
+ return nil, errors.Wrap(err, "error parsing address")
+ }
+ u.address = *newAddress
+ }
+
+ // promote the normal and broadcast tuples
+ u.addrTuple = u.address.AddrTuple
+ u.addrBroadcastTuple = u.address.AddrBroadcastTuple
+
+ // check for no broadcasting (loopback interface)
+ if u.addrBroadcastTuple == nil {
+ noBroadcast = true
+ } else if u.addrTuple == u.addrBroadcastTuple {
+ // old school broadcast address
+ u.addrBroadcastTuple = &AddressTuple[string, uint16]{"255.255.255.255", u.addrTuple.Right}
+ } else {
+ specialBroadcast = true
}
- u.address = udpAddr
- // TODO: we need to find a way to resolved broadcast
- u.selfBroadcastAddress = &net.UDPAddr{IP: net.IPv4(255, 255, 255, 255), Port: 47808}
}
- log.Debug().Msgf("address %s, broadcast %s", u.address, u.selfBroadcastAddress)
+ log.Debug().Msgf("address: %v", u.address)
+ log.Debug().Msgf("addrTuple: %v", u.addrTuple)
+ log.Debug().Msgf("addrBroadcastTuple: %v", u.addrBroadcastTuple)
+ //log.Debug().Msgf("route_aware: %v", settings.RouteAware)
// create and bind direct address
var err error
@@ -106,7 +129,7 @@ func NewUDPMultiplexer(address net.Addr, noBroadcast bool) (*UDPMultiplexer, err
if err != nil {
return nil, errors.Wrap(err, "error creating multiplex client")
}
- u.directPort, err = NewUDPDirector(u.address, nil, nil, nil, nil)
+ u.directPort, err = NewUDPDirector(u.addrTuple, nil, nil, nil, nil)
if err := bind(u.direct, u.directPort); err != nil {
return nil, errors.Wrap(err, "error binding ports")
}
@@ -117,7 +140,8 @@ func NewUDPMultiplexer(address net.Addr, noBroadcast bool) (*UDPMultiplexer, err
if err != nil {
return nil, errors.Wrap(err, "error creating broadcast multiplex client")
}
- u.broadcastPort, err = NewUDPDirector(u.selfBroadcastAddress, nil, nil, nil, nil)
+ reuse := true
+ u.broadcastPort, err = NewUDPDirector(u.addrBroadcastTuple, nil, &reuse, nil, nil)
if err := bind(u.direct, u.directPort); err != nil {
return nil, errors.Wrap(err, "error binding ports")
}
@@ -136,15 +160,57 @@ func NewUDPMultiplexer(address net.Addr, noBroadcast bool) (*UDPMultiplexer, err
}
func (m *UDPMultiplexer) Close() error {
- panic("implement me")
+ log.Debug().Msg("Close")
+
+ // pass along the close to the director(s)
+ m.directPort.Close()
+ if m.broadcastPort != nil {
+ m.broadcastPort.Close()
+ }
+ return nil
}
-func (m *UDPMultiplexer) Confirmation(pdu spi.Message) error {
- panic("implement me")
+func (m *UDPMultiplexer) Indication(server *_MultiplexServer, pdu _PDU) error {
+ log.Debug().Msgf("Indication %v\n%v", server, pdu)
+
+ pduDestination := pdu.GetPDUDestination()
+
+ // broadcast message
+ var dest Address
+ if pduDestination.AddrType == LOCAL_BROADCAST_ADDRESS {
+ // interface might not support broadcasts
+ if m.addrBroadcastTuple == nil {
+ return nil
+ }
+
+ address, err := NewAddress(*m.addrBroadcastTuple)
+ if err != nil {
+ return errors.Wrap(err, "error getting address from tuple")
+ }
+ dest = *address
+ log.Debug().Msgf("requesting local broadcast: %v", dest)
+ } else if pduDestination.AddrType == LOCAL_STATION_ADDRESS {
+ dest = pduDestination
+ } else {
+ return errors.New("invalid destination address type")
+ }
+
+ return m.directPort.Indication(NewPDUFromPDU(pdu, WithPDUDestination(dest)))
}
-func (m *UDPMultiplexer) Indication(pdu spi.Message) error {
- panic("implement me")
+func (m *UDPMultiplexer) Confirmation(client *_MultiplexClient, pdu _PDU) error {
+ log.Debug().Msgf("Confirmation %v\n%v", client, pdu)
+ log.Debug().Msgf("client address: %v", client.multiplexer.address)
+
+ // if this came from ourselves, dump it
+ pduSource := pdu.GetPDUSource()
+ if pduSource.Equals(m.address) {
+ log.Debug().Msg("from us")
+ return nil
+ }
+
+ // TODO: it is getting to messy, we need to solve the source destination topic
+ return nil
}
type AnnexJCodec struct {
@@ -168,12 +234,12 @@ func NewAnnexJCodec(cid *int, sid *int) (*AnnexJCodec, error) {
return a, nil
}
-func (b *AnnexJCodec) Indication(apdu spi.Message) error {
- panic("we need to implement this with generics as we handle npdu not apdu here")
+func (b *AnnexJCodec) Indication(apdu _PDU) error {
+ panic("not implemented yet")
}
-func (b *AnnexJCodec) Confirmation(apdu spi.Message) error {
- panic("we need to implement this with generics as we handle npdu not apdu here")
+func (b *AnnexJCodec) Confirmation(apdu _PDU) error {
+ panic("not implemented yet")
}
type _BIPSAP interface {
@@ -198,22 +264,14 @@ func NewBIPSAP(sapID *int, rootStruct _BIPSAP) (*BIPSAP, error) {
return b, nil
}
-func (b *BIPSAP) SapIndication(pdu spi.Message) error {
- // TODO: extract from somewhere
- var pduDestination []byte
- panic("we need pduDestination")
- log.Debug().Msgf("SapIndication\n%s\n%s", pdu, pduDestination)
- // TODO: what to do with the destination?
+func (b *BIPSAP) SapIndication(pdu _PDU) error {
+ log.Debug().Msgf("SapIndication\n%ss", pdu)
// this is a request initiated by the ASE, send this downstream
return b.rootStruct.Request(pdu)
}
-func (b *BIPSAP) SapConfirmation(pdu spi.Message) error {
- // TODO: extract from somewhere
- var pduDestination []byte
- panic("we need pduDestination")
- log.Debug().Msgf("SapConfirmation\n%s\n%s", pdu, pduDestination)
- // TODO: what to do with the destination?
+func (b *BIPSAP) SapConfirmation(pdu _PDU) error {
+ log.Debug().Msgf("SapConfirmation\n%s", pdu)
// this is a response from the ASE, send this downstream
return b.rootStruct.Request(pdu)
}
@@ -245,10 +303,10 @@ func NewBIPSimple(sapID *int, cid *int, sid *int) (*BIPSimple, error) {
return b, nil
}
-func (b *BIPSimple) Indication(apdu spi.Message) error {
- panic("we need to implement this with generics as we handle npdu not apdu here")
+func (b *BIPSimple) Indication(apdu _PDU) error {
+ panic("not implemented yet")
}
-func (b *BIPSimple) Response(apdu spi.Message) error {
- panic("we need to implement this with generics as we handle npdu not apdu here")
+func (b *BIPSimple) Response(apdu _PDU) error {
+ panic("not implemented yet")
}
diff --git a/plc4go/internal/bacnetip/CommunicationsModule.go b/plc4go/internal/bacnetip/CommunicationsModule.go
index 46b34c9aa1..38c8f6c041 100644
--- a/plc4go/internal/bacnetip/CommunicationsModule.go
+++ b/plc4go/internal/bacnetip/CommunicationsModule.go
@@ -20,7 +20,7 @@
package bacnetip
import (
- "github.com/apache/plc4x/plc4go/spi"
+ "fmt"
"github.com/pkg/errors"
"github.com/rs/zerolog/log"
)
@@ -40,11 +40,23 @@ func init() {
elementMap = make(map[int]*ApplicationServiceElement)
}
+// TODO: implement me
+type _PCI struct {
+ pduUserData interface{}
+ pduSource Address
+ pduDestination Address
+}
+
+func _New_PCI(pduUserData interface{}, pduSource Address, pduDestination Address) *_PCI {
+ return &_PCI{pduUserData, pduSource, pduDestination}
+}
+
// _Client is an interface used for documentation
type _Client interface {
- Request(pdu spi.Message) error
- Confirmation(pdu spi.Message) error
+ Request(pdu _PDU) error
+ Confirmation(pdu _PDU) error
_setClientPeer(server _Server)
+ getClientId() *int
}
// Client is an "abstract" struct which is used in another struct as delegate
@@ -78,7 +90,7 @@ func NewClient(cid *int, rootStruct _Client) (*Client, error) {
return c, nil
}
-func (c *Client) Request(pdu spi.Message) error {
+func (c *Client) Request(pdu _PDU) error {
log.Debug().Msgf("request\n%s", pdu)
if c.clientPeer == nil {
@@ -87,7 +99,7 @@ func (c *Client) Request(pdu spi.Message) error {
return c.clientPeer.Indication(pdu)
}
-func (c *Client) Confirmation(spi.Message) error {
+func (c *Client) Confirmation(_PDU) error {
panic("this should be implemented by outer struct")
}
@@ -95,11 +107,24 @@ func (c *Client) _setClientPeer(server _Server) {
c.clientPeer = server
}
+func (c *Client) getClientId() *int {
+ return c.clientID
+}
+
+func (c *Client) String() string {
+ clientPeer := ""
+ if c.clientPeer != nil {
+ clientPeer = fmt.Sprintf(" clientPeerId: %d", c.clientPeer.getServerId())
+ }
+ return fmt.Sprintf("Client(cid:%d)%s", c.clientID, clientPeer)
+}
+
// _Server is an interface used for documentation
type _Server interface {
- Indication(pdu spi.Message) error
- Response(pdu spi.Message) error
+ Indication(pdu _PDU) error
+ Response(pdu _PDU) error
_setServerPeer(serverPeer _Client)
+ getServerId() *int
}
// Server is an "abstract" struct which is used in another struct as delegate
@@ -133,11 +158,11 @@ func NewServer(sid *int, rootStruct _Server) (*Server, error) {
return s, nil
}
-func (s *Server) Indication(spi.Message) error {
+func (s *Server) Indication(_PDU) error {
panic("this should be implemented by outer struct")
}
-func (s *Server) Response(pdu spi.Message) error {
+func (s *Server) Response(pdu _PDU) error {
log.Debug().Msgf("response\n%s", pdu)
if s.serverPeer == nil {
@@ -150,12 +175,24 @@ func (s *Server) _setServerPeer(serverPeer _Client) {
s.serverPeer = serverPeer
}
+func (s *Server) getServerId() *int {
+ return s.serverID
+}
+
+func (s *Server) String() string {
+ serverPeer := ""
+ if s.serverPeer != nil {
+ serverPeer = fmt.Sprintf(" serverPeerId: %d", s.serverPeer.getClientId())
+ }
+ return fmt.Sprintf("Server(cid:%d)%s", s.serverID, serverPeer)
+}
+
// _ServiceAccessPoint is a interface used for documentation
type _ServiceAccessPoint interface {
- SapConfirmation(pdu spi.Message) error
- SapRequest(pdu spi.Message) error
- SapIndication(pdu spi.Message) error
- SapResponse(pdu spi.Message) error
+ SapConfirmation(pdu _PDU) error
+ SapRequest(pdu _PDU) error
+ SapIndication(pdu _PDU) error
+ SapResponse(pdu _PDU) error
_setServiceElement(serviceElement _ApplicationServiceElement)
}
@@ -189,7 +226,7 @@ func NewServiceAccessPoint(sapID *int, rootStruct _ServiceAccessPoint) (*Service
return s, nil
}
-func (s *ServiceAccessPoint) SapRequest(pdu spi.Message) error {
+func (s *ServiceAccessPoint) SapRequest(pdu _PDU) error {
log.Debug().Msgf("SapRequest(%d)\n%s", s.serviceID, pdu)
if s.serviceElement == nil {
@@ -198,11 +235,11 @@ func (s *ServiceAccessPoint) SapRequest(pdu spi.Message) error {
return s.serviceElement.Indication(pdu)
}
-func (s *ServiceAccessPoint) SapIndication(spi.Message) error {
+func (s *ServiceAccessPoint) SapIndication(_PDU) error {
panic("this should be implemented by outer struct")
}
-func (s *ServiceAccessPoint) SapResponse(pdu spi.Message) error {
+func (s *ServiceAccessPoint) SapResponse(pdu _PDU) error {
log.Debug().Msgf("SapResponse(%d)\n%s", s.serviceID, pdu)
if s.serviceElement == nil {
@@ -211,7 +248,7 @@ func (s *ServiceAccessPoint) SapResponse(pdu spi.Message) error {
return s.serviceElement.Confirmation(pdu)
}
-func (s *ServiceAccessPoint) SapConfirmation(spi.Message) error {
+func (s *ServiceAccessPoint) SapConfirmation(_PDU) error {
panic("this should be implemented by outer struct")
}
@@ -221,10 +258,10 @@ func (s *ServiceAccessPoint) _setServiceElement(serviceElement _ApplicationServi
// _ApplicationServiceElement is a interface used for documentation
type _ApplicationServiceElement interface {
- Request(pdu spi.Message) error
- Indication(pdu spi.Message) error
- Response(pdu spi.Message) error
- Confirmation(pdu spi.Message) error
+ Request(pdu _PDU) error
+ Indication(pdu _PDU) error
+ Response(pdu _PDU) error
+ Confirmation(pdu _PDU) error
_setElementService(elementService _ServiceAccessPoint)
}
@@ -259,7 +296,7 @@ func NewApplicationServiceElement(aseID *int, rootStruct _ApplicationServiceElem
return a, nil
}
-func (a *ApplicationServiceElement) Request(pdu spi.Message) error {
+func (a *ApplicationServiceElement) Request(pdu _PDU) error {
log.Debug().Msgf("Request\n%s", pdu)
if a.elementService == nil {
@@ -269,11 +306,11 @@ func (a *ApplicationServiceElement) Request(pdu spi.Message) error {
return a.elementService.SapIndication(pdu)
}
-func (a *ApplicationServiceElement) Indication(spi.Message) error {
+func (a *ApplicationServiceElement) Indication(_PDU) error {
panic("this should be implemented by outer struct")
}
-func (a *ApplicationServiceElement) Response(pdu spi.Message) error {
+func (a *ApplicationServiceElement) Response(pdu _PDU) error {
log.Debug().Msgf("Response\n%s", pdu)
if a.elementService == nil {
@@ -283,7 +320,7 @@ func (a *ApplicationServiceElement) Response(pdu spi.Message) error {
return a.elementService.SapConfirmation(pdu)
}
-func (a *ApplicationServiceElement) Confirmation(spi.Message) error {
+func (a *ApplicationServiceElement) Confirmation(_PDU) error {
panic("this should be implemented by outer struct")
}
@@ -367,9 +404,9 @@ func bind(args ...interface{}) error {
// go through the argument pairs
for i := 0; i < len(args)-1; i++ {
client := args[i]
- log.Debug().Msgf("client %v", client)
+ log.Debug().Msgf("client %s", client)
server := args[i+1]
- log.Debug().Msgf("server %v", server)
+ log.Debug().Msgf("server %s", server)
// make sure we're binding clients and servers
clientCast, okClient := client.(_Client)
diff --git a/plc4go/internal/bacnetip/Driver.go b/plc4go/internal/bacnetip/Driver.go
index dfccafe687..0fef164050 100644
--- a/plc4go/internal/bacnetip/Driver.go
+++ b/plc4go/internal/bacnetip/Driver.go
@@ -50,8 +50,10 @@ type Driver struct {
func NewDriver() plc4go.PlcDriver {
return &Driver{
- DefaultDriver: _default.NewDefaultDriver("bacnet-ip", "BACnet/IP", "udp", NewTagHandler()),
- applicationManager: ApplicationManager{},
+ DefaultDriver: _default.NewDefaultDriver("bacnet-ip", "BACnet/IP", "udp", NewTagHandler()),
+ applicationManager: ApplicationManager{
+ applications: map[string]*ApplicationLayerMessageCodec{},
+ },
tm: *spi.NewRequestTransactionManager(math.MaxInt),
awaitSetupComplete: true,
awaitDisconnectComplete: true,
diff --git a/plc4go/internal/bacnetip/IOCBModule.go b/plc4go/internal/bacnetip/IOCBModule.go
index e51fa8456c..5cdec6ae46 100644
--- a/plc4go/internal/bacnetip/IOCBModule.go
+++ b/plc4go/internal/bacnetip/IOCBModule.go
@@ -22,7 +22,6 @@ package bacnetip
import (
"container/heap"
"fmt"
- "github.com/apache/plc4x/plc4go/spi"
"github.com/apache/plc4x/plc4go/spi/plcerrors"
"github.com/apache/plc4x/plc4go/spi/utils"
"github.com/pkg/errors"
@@ -84,10 +83,10 @@ type _IOCB interface {
setIOController(ioController _IOController)
setIOState(newState IOCBState)
getIOState() IOCBState
- setIOResponse(msg spi.Message)
+ setIOResponse(msg _PDU)
Trigger()
setIOError(err error)
- getRequest() spi.Message
+ getRequest() _PDU
getDestination() net.Addr
getPriority() int
clearQueue()
@@ -99,10 +98,10 @@ var _identLock sync.Mutex
type IOCB struct {
ioID int
- request spi.Message
+ request _PDU
destination net.Addr
ioState IOCBState
- ioResponse spi.Message
+ ioResponse _PDU
ioError error
ioController _IOController
ioComplete sync.Cond
@@ -114,7 +113,7 @@ type IOCB struct {
priority int
}
-func NewIOCB(request spi.Message, destination net.Addr) (*IOCB, error) {
+func NewIOCB(request _PDU, destination net.Addr) (*IOCB, error) {
// lock the identity sequence number
_identLock.Lock()
@@ -194,7 +193,7 @@ func (i *IOCB) Trigger() {
// Complete Called to complete a transaction, usually when ProcessIO has shipped the IOCB off to some other thread or
// function.
-func (i *IOCB) Complete(apdu spi.Message) error {
+func (i *IOCB) Complete(apdu _PDU) error {
log.Debug().Msgf("Complete(%d)\n%s", i.ioID, apdu)
if i.ioController != nil {
@@ -257,7 +256,7 @@ func (i *IOCB) getIOState() IOCBState {
return i.ioState
}
-func (i *IOCB) setIOResponse(msg spi.Message) {
+func (i *IOCB) setIOResponse(msg _PDU) {
i.ioResponse = msg
}
@@ -265,7 +264,7 @@ func (i *IOCB) setIOError(err error) {
i.ioError = err
}
-func (i *IOCB) getRequest() spi.Message {
+func (i *IOCB) getRequest() _PDU {
return i.request
}
@@ -434,7 +433,7 @@ func (i *IOQueue) Abort(err error) {
type _IOController interface {
Abort(err error) error
ProcessIO(iocb _IOCB) error
- CompleteIO(iocb _IOCB, pdu spi.Message) error
+ CompleteIO(iocb _IOCB, pdu _PDU) error
AbortIO(iocb _IOCB, err error) error
}
@@ -500,7 +499,7 @@ func (i *IOController) ActiveIO(iocb _IOCB) error {
}
// CompleteIO Called by a handler to return data to the client
-func (i *IOController) CompleteIO(iocb _IOCB, apdu spi.Message) error {
+func (i *IOController) CompleteIO(iocb _IOCB, apdu _PDU) error {
log.Debug().Msgf("CompleteIO %s\n%s", iocb, apdu)
// if it completed, leave it alone
@@ -579,11 +578,11 @@ func NewIOQController(name string) (*IOQController, error) {
type SieveQueue struct {
*IOQController
- requestFn func(apdu spi.Message)
+ requestFn func(apdu _PDU)
address net.Addr
}
-func NewSieveQueue(fn func(apdu spi.Message), address net.Addr) (*SieveQueue, error) {
+func NewSieveQueue(fn func(apdu _PDU), address net.Addr) (*SieveQueue, error) {
s := &SieveQueue{}
var err error
s.IOQController, err = NewIOQController(address.String())
diff --git a/plc4go/internal/bacnetip/MessageCodec.go b/plc4go/internal/bacnetip/MessageCodec.go
index 45b51af713..d76e38e374 100644
--- a/plc4go/internal/bacnetip/MessageCodec.go
+++ b/plc4go/internal/bacnetip/MessageCodec.go
@@ -55,7 +55,11 @@ func NewApplicationLayerMessageCodec(udpTransport *udp.Transport, transportUrl u
localAddress: localAddress,
remoteAddress: remoteAddress,
}
- application, err := NewBIPSimpleApplication(&local.LocalDeviceObject{}, localAddress, &a.deviceInfoCache, nil)
+ address, err := NewAddress(localAddress)
+ if err != nil {
+ return nil, errors.Wrap(err, "error creating address")
+ }
+ application, err := NewBIPSimpleApplication(&local.LocalDeviceObject{}, *address, &a.deviceInfoCache, nil)
if err != nil {
return nil, err
}
@@ -88,7 +92,11 @@ func (m *ApplicationLayerMessageCodec) IsRunning() bool {
}
func (m *ApplicationLayerMessageCodec) Send(message spi.Message) error {
- iocb, err := NewIOCB(message.(model.APDU), m.remoteAddress)
+ address, err2 := NewAddress(m.remoteAddress)
+ if err2 != nil {
+ panic(err2)
+ }
+ iocb, err := NewIOCB(NewPDU(message, WithPDUDestination(*address)), m.remoteAddress)
if err != nil {
return errors.Wrap(err, "error creating IOCB")
}
diff --git a/plc4go/internal/bacnetip/NetworkService.go b/plc4go/internal/bacnetip/NetworkService.go
index ce196f858d..5017f05063 100644
--- a/plc4go/internal/bacnetip/NetworkService.go
+++ b/plc4go/internal/bacnetip/NetworkService.go
@@ -20,10 +20,9 @@
package bacnetip
import (
- "github.com/apache/plc4x/plc4go/spi"
+ "fmt"
"github.com/pkg/errors"
"github.com/rs/zerolog/log"
- "net"
)
// TODO: implement me
@@ -31,11 +30,11 @@ type NetworkAdapter struct {
*Client
adapterSAP *NetworkServiceAccessPoint
adapterNet interface{}
- adapterAddr net.Addr
+ adapterAddr *Address
adapterNetConfigured *int
}
-func NewNetworkAdapter(sap *NetworkServiceAccessPoint, net interface{}, addr net.Addr, cid *int) (*NetworkAdapter, error) {
+func NewNetworkAdapter(sap *NetworkServiceAccessPoint, net interface{}, addr *Address, cid *int) (*NetworkAdapter, error) {
n := &NetworkAdapter{
adapterSAP: sap,
adapterNet: net,
@@ -55,7 +54,7 @@ func NewNetworkAdapter(sap *NetworkServiceAccessPoint, net interface{}, addr net
}
// Confirmation Decode upstream PDUs and pass them up to the service access point.
-func (n *NetworkAdapter) Confirmation(npdu spi.Message) error {
+func (n *NetworkAdapter) Confirmation(npdu _PDU) error {
log.Debug().Msgf("confirmation\n%s\n%s", npdu, n.adapterNet)
// TODO: we need generics otherwise this won't work at all here
@@ -63,7 +62,7 @@ func (n *NetworkAdapter) Confirmation(npdu spi.Message) error {
}
// ProcessNPDU Encode NPDUs from the service access point and send them downstream.
-func (n *NetworkAdapter) ProcessNPDU(npdu spi.Message) error {
+func (n *NetworkAdapter) ProcessNPDU(npdu _PDU) error {
log.Debug().Msgf("ProcessNPDU\n%s\n(net=%s)", npdu, n.adapterNet)
return n.Request(npdu)
}
@@ -74,7 +73,7 @@ type NetworkServiceAccessPoint struct {
adapters map[string]*NetworkAdapter
routerInfoCache interface{}
pendingNets map[string]interface{}
- localAdapter interface{}
+ localAdapter *NetworkAdapter
}
func NewNetworkServiceAccessPoint(routerInfoCache interface{}, sapID *int, sid *int) (*NetworkServiceAccessPoint, error) {
@@ -104,8 +103,57 @@ func NewNetworkServiceAccessPoint(routerInfoCache interface{}, sapID *int, sid *
return n, nil
}
-func (n *NetworkServiceAccessPoint) bind(server _Server, net interface{}, address interface{}) error {
- panic("not implemented yet")
+/* bind creates a network adapter object and bind.
+
+ bind(s, None, None)
+ Called for simple applications, local network unknown, no specific
+ address, APDUs sent upstream
+
+ bind(s, net, None)
+ Called for routers, bind to the network, (optionally?) drop APDUs
+
+ bind(s, None, address)
+ Called for applications or routers, bind to the network (to be
+ discovered), send up APDUs with a metching address
+
+ bind(s, net, address)
+ Called for applications or routers, bind to the network, send up
+ APDUs with a metching address.
+*/
+func (n *NetworkServiceAccessPoint) bind(server _Server, net interface{}, address *Address) error {
+ log.Debug().Msgf("bind %v net=%v address=%v", server, net, address)
+
+ netKey := fmt.Sprintf("%v", net)
+ // make sure this hasn't already been called with this network
+ if _, ok := n.adapters[netKey]; ok {
+ return errors.Errorf("Allready bound: %v", net)
+ }
+ // create an adapter object, add it to our map
+ adapter, err := NewNetworkAdapter(n, net, address, nil)
+ if err != nil {
+ return errors.Wrap(err, "error creating adapter")
+ }
+ n.adapters[netKey] = adapter
+ log.Debug().Msgf("adapter: %v, %v", netKey, adapter)
+
+ // if the address was given, make it the "local" one
+ if address != nil {
+ log.Debug().Msg("setting local adapter")
+ n.localAdapter = adapter
+ }
+
+ // if the local adapter isn't set yet, make it the first one, and can
+ // be overridden by a subsequent call if the address is specified
+ if n.localAdapter == nil {
+ log.Debug().Msg("default local adapter")
+ n.localAdapter = adapter
+ }
+
+ if n.localAdapter.adapterAddr == nil {
+ log.Debug().Msg("no local address")
+ }
+
+ return bind(adapter, server)
}
func (n *NetworkServiceAccessPoint) UpdateRouterReference() error {
@@ -116,27 +164,19 @@ func (n *NetworkServiceAccessPoint) DeleteRouterReference() error {
panic("not implemented yet")
}
-func (n *NetworkServiceAccessPoint) Indication(npdu spi.Message) error {
+func (n *NetworkServiceAccessPoint) Indication(npdu _PDU) error {
panic("not implemented yet")
}
-func (n *NetworkServiceAccessPoint) ProcessNPDU(npdu spi.Message) error {
+func (n *NetworkServiceAccessPoint) ProcessNPDU(npdu _PDU) error {
panic("not implemented yet")
}
-func (n *NetworkServiceAccessPoint) SapIndication(npdu spi.Message) error {
- // TODO: extract from somewhere
- var pduDestination []byte
- panic("we need pduDestination")
- _ = pduDestination
+func (n *NetworkServiceAccessPoint) SapIndication(npdu _PDU) error {
panic("not implemented yet")
}
-func (n *NetworkServiceAccessPoint) SapConfirmation(npdu spi.Message) error {
- // TODO: extract from somewhere
- var pduDestination []byte
- panic("we need pduDestination")
- _ = pduDestination
+func (n *NetworkServiceAccessPoint) SapConfirmation(npdu _PDU) error {
panic("not implemented yet")
}
diff --git a/plc4go/internal/bacnetip/PDU.go b/plc4go/internal/bacnetip/PDU.go
new file mode 100644
index 0000000000..5acf571e7b
--- /dev/null
+++ b/plc4go/internal/bacnetip/PDU.go
@@ -0,0 +1,482 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package bacnetip
+
+import (
+ "encoding/binary"
+ "fmt"
+ readWriteModel "github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model"
+ "github.com/apache/plc4x/plc4go/spi"
+ "github.com/pkg/errors"
+ "github.com/rs/zerolog/log"
+ "net"
+ "reflect"
+)
+
+type AddressType int
+
+const (
+ NULL_ADDRESS AddressType = iota
+ LOCAL_BROADCAST_ADDRESS
+ LOCAL_STATION_ADDRESS
+ REMOTE_BROADCAST_ADDRESS
+ REMOTE_STATION_ADDRESS
+ GLOBAL_BROADCAST_ADDRESS
+)
+
+func (a AddressType) String() string {
+ switch a {
+ case NULL_ADDRESS:
+ return "NULL_ADDRESS"
+ case LOCAL_BROADCAST_ADDRESS:
+ return "LOCAL_BROADCAST_ADDRESS"
+ case LOCAL_STATION_ADDRESS:
+ return "LOCAL_STATION_ADDRESS"
+ case REMOTE_BROADCAST_ADDRESS:
+ return "REMOTE_BROADCAST_ADDRESS"
+ case REMOTE_STATION_ADDRESS:
+ return "REMOTE_STATION_ADDRESS"
+ case GLOBAL_BROADCAST_ADDRESS:
+ return "GLOBAL_BROADCAST_ADDRESS"
+ default:
+ return "Unknown"
+ }
+}
+
+type AddressTuple[L any, R any] struct {
+ Left L
+ Right R
+}
+
+type Address struct {
+ AddrType AddressType
+ AddrNet *uint16
+ AddrAddress []byte
+ AddrLen *uint32
+ AddrRoute *uint32
+
+ AddrIP *uint32
+ AddrMask *uint32
+ AddrHost *uint32
+ AddrSubnet *uint32
+ AddrPort *uint16
+ AddrTuple *AddressTuple[string, uint16]
+ AddrBroadcastTuple *AddressTuple[string, uint16]
+}
+
+func NewAddress(args ...interface{}) (*Address, error) {
+ log.Debug().Interface("args", args).Msg("NewAddress")
+ a := &Address{}
+ a.AddrNet = nil
+ a.AddrAddress = nil
+ a.AddrLen = nil
+ a.AddrRoute = nil
+
+ switch len(args) {
+ case 1:
+ if err := a.decodeAddress(args[0]); err != nil {
+ return nil, errors.Wrap(err, "decodeAddress")
+ }
+ case 2:
+ if err := a.decodeAddress(args[1]); err != nil {
+ return nil, errors.Wrap(err, "decodeAddress")
+ }
+ switch a.AddrType {
+ case LOCAL_STATION_ADDRESS:
+ a.AddrType = REMOTE_STATION_ADDRESS
+ var net = (args[0]).(uint16)
+ a.AddrNet = &net
+ case LOCAL_BROADCAST_ADDRESS:
+ a.AddrType = REMOTE_BROADCAST_ADDRESS
+ var net = (args[0]).(uint16)
+ a.AddrNet = &net
+ default:
+ return nil, errors.New("unrecognized address ctor form")
+ }
+ }
+ return a, nil
+}
+
+// decodeAddress Initialize the address from a string. Lots of different forms are supported
+func (a *Address) decodeAddress(addr interface{}) error {
+ log.Debug().Msgf("decodeAddress %v (%T)", addr, addr)
+
+ // start out assuming this is a local station and didn't get routed
+ a.AddrType = LOCAL_STATION_ADDRESS
+ a.AddrNet = nil
+ a.AddrAddress = nil
+ a.AddrLen = nil
+ a.AddrRoute = nil
+
+ switch {
+ case addr == "*":
+ log.Debug().Msg("localBroadcast")
+ a.AddrType = LOCAL_BROADCAST_ADDRESS
+ case addr == "*:*":
+ log.Debug().Msg("globalBroadcast")
+ a.AddrType = GLOBAL_BROADCAST_ADDRESS
+ default:
+ switch addr := addr.(type) {
+ case net.Addr:
+ // TODO: hacked in udp support
+ udpAddr := addr.(*net.UDPAddr)
+ a.AddrAddress = udpAddr.IP
+ length := uint32(len(a.AddrAddress))
+ a.AddrLen = &length
+ port := uint16(udpAddr.Port)
+ a.AddrPort = &port
+ addr.String()
+ case int:
+ log.Debug().Msg("int")
+ if addr < 0 || addr > 255 {
+ return errors.New("address out of range")
+ }
+ a.AddrAddress = []byte{byte(addr)}
+ length := uint32(1)
+ a.AddrLen = &length
+ case []byte:
+ log.Debug().Msg("byte array")
+ a.AddrAddress = addr
+ length := uint32(len(addr))
+ a.AddrLen = &length
+
+ if *a.AddrLen == 6 {
+ ip := ipv4ToUint32(addr[:4])
+ a.AddrIP = &ip
+ mask := uint32((1 << 32) - 1)
+ a.AddrMask = &mask
+ host := *a.AddrIP & ^(*a.AddrMask)
+ a.AddrHost = &host
+ subnet := *a.AddrIP & *a.AddrMask
+ a.AddrSubnet = &subnet
+ port := portToUint16(addr[4:])
+ a.AddrPort = &port
+
+ a.AddrTuple = &AddressTuple[string, uint16]{uint32ToIpv4(*a.AddrIP).String(), *a.AddrPort}
+ a.AddrBroadcastTuple = &AddressTuple[string, uint16]{"255.255.255.255", *a.AddrPort}
+ }
+ case string:
+ log.Debug().Msg("str")
+
+ panic("parsing not yet ported")
+ case AddressTuple[string, uint16]:
+ uaddr, port := addr.Left, addr.Right
+ a.AddrPort = &port
+
+ var addrstr []byte
+ if uaddr == "" {
+ // when ('', n) is passed it is the local host address, but that could be more than one on a multi homed machine,
+ // the empty string # means "any".
+ addrstr = make([]byte, 4)
+ } else {
+ addrstr = net.ParseIP(uaddr)
+ }
+ a.AddrTuple = &AddressTuple[string, uint16]{uaddr, *a.AddrPort}
+ log.Debug().Msgf("addrstr: %s", addrstr)
+
+ ip := ipv4ToUint32(addrstr)
+ a.AddrIP = &ip
+ mask := uint32(0xFFFFFFFF)
+ a.AddrMask = &mask
+ host := uint32(0)
+ a.AddrHost = &host
+ subnet := uint32(0)
+ a.AddrSubnet = &subnet
+ a.AddrBroadcastTuple = a.AddrTuple
+
+ a.AddrAddress = append(addrstr, uint16ToPort(*a.AddrPort)...)
+ length := uint32(6)
+ a.AddrLen = &length
+ case AddressTuple[int, uint16]:
+ uaddr, port := addr.Left, addr.Right
+ a.AddrPort = &port
+
+ addrstr := uint32ToIpv4(uint32(uaddr))
+ a.AddrTuple = &AddressTuple[string, uint16]{addrstr.String(), *a.AddrPort}
+ log.Debug().Msgf("addrstr: %s", addrstr)
+
+ ip := ipv4ToUint32(addrstr)
+ a.AddrIP = &ip
+ mask := uint32(0xFFFFFFFF)
+ a.AddrMask = &mask
+ host := uint32(0)
+ a.AddrHost = &host
+ subnet := uint32(0)
+ a.AddrSubnet = &subnet
+ a.AddrBroadcastTuple = a.AddrTuple
+
+ a.AddrAddress = append(addrstr, uint16ToPort(*a.AddrPort)...)
+ length := uint32(6)
+ a.AddrLen = &length
+ default:
+ return errors.Errorf("integer, string or tuple required (Actual %T)", addr)
+ }
+ }
+ return nil
+}
+
+func (a *Address) Equals(other interface{}) bool {
+ if a == nil && other == nil {
+ return true
+ } else if a == nil && other != nil {
+ return false
+ }
+ switch other := other.(type) {
+ case *Address:
+ if a == other {
+ return true
+ }
+ // TODO: don't use reflect here
+ return reflect.DeepEqual(a, other)
+ case Address:
+ // TODO: don't use reflect here
+ return reflect.DeepEqual(*a, other)
+ default:
+ return false
+ }
+}
+
+func (a *Address) String() string {
+ if a == nil {
+ return "<nil>"
+ }
+ return fmt.Sprintf("Address{AddrType: %s, AddrNet: %d, AddrAddress: %x, AddrLen: %d, AddrRoute: %d, AddrIP: %d, AddrMask: %d, AddrHost: %d, AddrSubnet: %d, AddrPort: %d, AddrTuple: %v, AddrBroadcastTuple: %v}", a.AddrType, a.AddrNet, a.AddrAddress, a.AddrLen, a.AddrRoute, a.AddrIP, a.AddrMask, a.AddrHost, a.AddrSubnet, a.AddrPort, a.AddrTuple, a.AddrBroadcastTuple)
+}
+
+func portToUint16(port []byte) uint16 {
+ switch len(port) {
+ case 2:
+ default:
+ panic("port must be 2 bytes")
+ }
+ return binary.BigEndian.Uint16(port)
+}
+
+func uint16ToPort(number uint16) []byte {
+ port := make([]byte, 2)
+ binary.BigEndian.PutUint16(port, number)
+ return port
+}
+
+func ipv4ToUint32(ip net.IP) uint32 {
+ switch len(ip) {
+ case 4:
+ default:
+ panic("ip must be either 4 bytes")
+ }
+ return binary.BigEndian.Uint32(ip)
+}
+
+func uint32ToIpv4(number uint32) net.IP {
+ ipv4 := make(net.IP, 4)
+ binary.BigEndian.PutUint32(ipv4, number)
+ return ipv4
+}
+
+type LocalStation struct {
+ Address
+}
+
+func NewLocalStation(addr interface{}, route *uint32) (*LocalStation, error) {
+ l := &LocalStation{}
+ l.AddrType = LOCAL_STATION_ADDRESS
+ l.AddrRoute = route
+
+ switch addr := addr.(type) {
+ case int:
+ if addr < 0 || addr > 255 {
+ return nil, errors.New("address out of range")
+ }
+ l.AddrAddress = []byte{byte(addr)}
+ length := uint32(1)
+ l.AddrLen = &length
+ case []byte:
+ log.Debug().Msg("bytearray")
+ l.AddrAddress = addr
+ length := uint32(len(addr))
+ l.AddrLen = &length
+ default:
+ return nil, errors.New("integer or byte array required")
+ }
+ return l, nil
+}
+
+type RemoteStation struct {
+ Address
+}
+
+func NewRemoteStation(net *uint16, addr interface{}, route *uint32) (*RemoteStation, error) {
+ l := &RemoteStation{}
+ l.AddrType = REMOTE_STATION_ADDRESS
+ l.AddrNet = net
+ l.AddrRoute = route
+
+ switch addr := addr.(type) {
+ case int:
+ if addr < 0 || addr > 255 {
+ return nil, errors.New("address out of range")
+ }
+ l.AddrAddress = []byte{byte(addr)}
+ length := uint32(1)
+ l.AddrLen = &length
+ case []byte:
+ log.Debug().Msg("bytearray")
+ l.AddrAddress = addr
+ length := uint32(len(addr))
+ l.AddrLen = &length
+ default:
+ return nil, errors.New("integer or byte array required")
+ }
+ return l, nil
+}
+
+type LocalBroadcast struct {
+ Address
+}
+
+func NewLocalBroadcast(route *uint32) (*LocalBroadcast, error) {
+ l := &LocalBroadcast{}
+ l.AddrType = LOCAL_BROADCAST_ADDRESS
+ l.AddrRoute = route
+ return l, nil
+}
+
+type RemoteBroadcast struct {
+ Address
+}
+
+func NewRemoteBroadcast(net *uint16, route *uint32) (*RemoteBroadcast, error) {
+ r := &RemoteBroadcast{}
+ r.AddrType = REMOTE_BROADCAST_ADDRESS
+ r.AddrNet = net
+ r.AddrRoute = route
+ return r, nil
+}
+
+type GlobalBroadcast struct {
+ Address
+}
+
+func NewGlobalBroadcast(route *uint32) (*GlobalBroadcast, error) {
+ g := &GlobalBroadcast{}
+ g.AddrType = GLOBAL_BROADCAST_ADDRESS
+ g.AddrRoute = route
+ return g, nil
+}
+
+type PCI struct {
+ *_PCI
+ expectingReply bool
+ networkPriority readWriteModel.NPDUNetworkPriority
+}
+
+func NewPCI(msg spi.Message, pduSource Address, pduDestination Address, expectingReply bool, networkPriority readWriteModel.NPDUNetworkPriority) *PCI {
+ return &PCI{
+ _New_PCI(msg, pduSource, pduDestination),
+ expectingReply,
+ networkPriority,
+ }
+}
+
+type _PDU interface {
+ spi.Message
+ GetPDUSource() Address
+ GetPDUDestination() Address
+ GetExpectingReply() bool
+ GetNetworkPriority() readWriteModel.NPDUNetworkPriority
+}
+
+type PDU struct {
+ spi.Message
+ *PCI
+}
+
+func NewPDU(msg spi.Message, pduOptions ...PDUOption) *PDU {
+ nullAddress, _ := NewAddress()
+ p := &PDU{
+ msg,
+ NewPCI(msg, *nullAddress, *nullAddress, false, readWriteModel.NPDUNetworkPriority_NORMAL_MESSAGE),
+ }
+ for _, option := range pduOptions {
+ option(p)
+ }
+ return p
+}
+
+func NewPDUFromPDU(pdu _PDU, pduOptions ...PDUOption) *PDU {
+ msg := pdu.(*PDU).Message
+ p := &PDU{
+ msg,
+ NewPCI(msg, pdu.GetPDUSource(), pdu.GetPDUDestination(), pdu.GetExpectingReply(), pdu.GetNetworkPriority()),
+ }
+ for _, option := range pduOptions {
+ option(p)
+ }
+ return p
+}
+
+func NewPDUWithAllOptions(msg spi.Message, pduSource Address, pduDestination Address, expectingReply bool, networkPriority readWriteModel.NPDUNetworkPriority) *PDU {
+ return &PDU{
+ msg,
+ NewPCI(msg, pduSource, pduDestination, expectingReply, networkPriority),
+ }
+}
+
+type PDUOption func(pdu *PDU)
+
+func WithPDUSource(pduSource Address) PDUOption {
+ return func(pdu *PDU) {
+ pdu.pduSource = pduSource
+ }
+}
+
+func WithPDUDestination(pduDestination Address) PDUOption {
+ return func(pdu *PDU) {
+ pdu.pduDestination = pduDestination
+ }
+}
+
+func WithPDUExpectingReply(expectingReply bool) PDUOption {
+ return func(pdu *PDU) {
+ pdu.expectingReply = expectingReply
+ }
+}
+
+func WithPDUNetworkPriority(networkPriority readWriteModel.NPDUNetworkPriority) PDUOption {
+ return func(pdu *PDU) {
+ pdu.networkPriority = networkPriority
+ }
+}
+
+func (p *PDU) GetPDUSource() Address {
+ return p.pduSource
+}
+
+func (p *PDU) GetPDUDestination() Address {
+ return p.pduDestination
+}
+
+func (p *PDU) GetExpectingReply() bool {
+ return p.expectingReply
+}
+
+func (p *PDU) GetNetworkPriority() readWriteModel.NPDUNetworkPriority {
+ return p.networkPriority
+}
diff --git a/plc4go/internal/bacnetip/UDPCommunicationsModule.go b/plc4go/internal/bacnetip/UDPCommunicationsModule.go
index aeed7863c6..43714a3ebd 100644
--- a/plc4go/internal/bacnetip/UDPCommunicationsModule.go
+++ b/plc4go/internal/bacnetip/UDPCommunicationsModule.go
@@ -21,7 +21,6 @@ package bacnetip
import (
"github.com/pkg/errors"
- "net"
)
type UDPActor struct {
@@ -37,7 +36,11 @@ type UDPDirector struct {
*ServiceAccessPoint
}
-func NewUDPDirector(address net.Addr, timeout *int, reuse *bool, sid *int, sapID *int) (*UDPDirector, error) {
+func (d *UDPDirector) Close() {
+
+}
+
+func NewUDPDirector(address *AddressTuple[string, uint16], timeout *int, reuse *bool, sid *int, sapID *int) (*UDPDirector, error) {
u := &UDPDirector{}
var err error
u.Server, err = NewServer(sid, u)
diff --git a/plc4go/internal/bacnetip/local/Device.go b/plc4go/internal/bacnetip/local/Device.go
index 752013a2ff..0069a7f1e8 100644
--- a/plc4go/internal/bacnetip/local/Device.go
+++ b/plc4go/internal/bacnetip/local/Device.go
@@ -20,6 +20,7 @@
package local
import (
+ "fmt"
readWriteModel "github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model"
)
@@ -34,3 +35,7 @@ type LocalDeviceObject struct {
ObjectName string
ObjectIdentifier string
}
+
+func (l *LocalDeviceObject) String() string {
+ return fmt.Sprintf("LocalDeviceObject{NumberOfAPDURetries: %v, APDUTimeout: %v, SegmentationSupported: %v, APDUSegmentTimeout: %v, MaxSegmentsAccepted: %v, MaximumApduLengthAccepted: %v, ObjectName: %v, ObjectIdentifier: %v}", l.NumberOfAPDURetries, l.APDUTimeout, l.SegmentationSupported, l.APDUSegmentTimeout, l.MaxSegmentsAccepted, l.MaximumApduLengthAccepted, l.ObjectName, l.ObjectIdentifier)
+}
diff --git a/plc4go/tests/drivers/tests/manual_bacnet_driver_test.go b/plc4go/tests/drivers/tests/manual_bacnet_driver_test.go
index a0c7478304..2955641864 100644
--- a/plc4go/tests/drivers/tests/manual_bacnet_driver_test.go
+++ b/plc4go/tests/drivers/tests/manual_bacnet_driver_test.go
@@ -22,14 +22,25 @@ package tests
import (
"github.com/apache/plc4x/plc4go/internal/bacnetip"
"github.com/apache/plc4x/plc4go/pkg/api"
+ "github.com/apache/plc4x/plc4go/pkg/api/config"
"github.com/apache/plc4x/plc4go/pkg/api/transports"
"github.com/apache/plc4x/plc4go/spi/testutils"
_ "github.com/apache/plc4x/plc4go/tests/initializetest"
+ "github.com/rs/zerolog"
+ "github.com/rs/zerolog/log"
+ "os"
"testing"
)
func TestManualBacnetDriver(t *testing.T) {
t.Skip()
+ log.Logger = log.
+ With().Caller().Logger().
+ Output(zerolog.ConsoleWriter{Out: os.Stderr}).
+ Level(zerolog.DebugLevel)
+ config.TraceTransactionManagerWorkers = false
+ config.TraceTransactionManagerTransactions = false
+ config.TraceDefaultMessageCodecWorker = false
connectionString := "bacnet-ip://192.168.178.101"
driverManager := plc4go.NewPlcDriverManager()