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/11 17:32:47 UTC
[plc4x] 01/02: feat(plc4go/bacnet): Client, Server, ServiceAccessPoint and ApplicationServiceElement
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 9b2e27f09fb8f6fa56604f35fd11e70d7b751e23
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Fri Nov 11 16:08:54 2022 +0100
feat(plc4go/bacnet): Client, Server, ServiceAccessPoint and ApplicationServiceElement
restructured code into own go files
---
...nsactionStateMachine.go => ApplicationLayer.go} | 581 +++----
plc4go/internal/bacnetip/CommunicationsModule.go | 391 +++++
plc4go/internal/bacnetip/Task.go | 44 +
.../internal/bacnetip/TransactionStateMachine.go | 1723 --------------------
.../bacnetip/TransactionStateMachine_test.go | 432 -----
5 files changed, 731 insertions(+), 2440 deletions(-)
diff --git a/plc4go/internal/bacnetip/TransactionStateMachine.go b/plc4go/internal/bacnetip/ApplicationLayer.go
similarity index 83%
copy from plc4go/internal/bacnetip/TransactionStateMachine.go
copy to plc4go/internal/bacnetip/ApplicationLayer.go
index c4132e3744..050c9ac96f 100644
--- a/plc4go/internal/bacnetip/TransactionStateMachine.go
+++ b/plc4go/internal/bacnetip/ApplicationLayer.go
@@ -21,170 +21,13 @@ package bacnetip
import (
"bytes"
- "context"
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"
"time"
)
-// TransactionStateMachine is the implementation of the bacnet transaction state machine
-type TransactionStateMachine struct {
- *MessageCodec
- deviceInventory *DeviceInventory
- retryCount int
- segmentRetryCount int
- duplicateCount int
- sentAllSegments bool
- lastSequenceNumber int
- initialSequenceNumber int
- actualWindowSize int
- proposeWindowSize int
- segmentTimer int
- RequestTimer int
-}
-
-func NewTransactionStateMachine(messageCodec *MessageCodec, deviceInventory *DeviceInventory) TransactionStateMachine {
- return TransactionStateMachine{
- MessageCodec: messageCodec,
- deviceInventory: deviceInventory,
- retryCount: 3,
- segmentRetryCount: 3,
- duplicateCount: 0,
- sentAllSegments: false,
- lastSequenceNumber: 0,
- initialSequenceNumber: 0,
- actualWindowSize: 0,
- proposeWindowSize: 2,
- segmentTimer: 1500,
- RequestTimer: 3000,
- }
-}
-
-func (t *TransactionStateMachine) GetCodec() spi.MessageCodec {
- return t
-}
-
-func (t *TransactionStateMachine) Send(message spi.Message) error {
- if handled, err := t.handleOutboundMessage(message); handled {
- return nil
- } else if err != nil {
- return errors.Wrap(err, "Error handling message")
- } else {
- return t.MessageCodec.Send(message)
- }
-}
-
-func (t *TransactionStateMachine) Expect(ctx context.Context, acceptsMessage spi.AcceptsMessage, handleMessage spi.HandleMessage, handleError spi.HandleError, ttl time.Duration) error {
- // TODO: detect overflow
- return t.MessageCodec.Expect(ctx, acceptsMessage, handleMessage, handleError, ttl)
-}
-
-func (t *TransactionStateMachine) SendRequest(ctx context.Context, message spi.Message, acceptsMessage spi.AcceptsMessage, handleMessage spi.HandleMessage, handleError spi.HandleError, ttl time.Duration) error {
- // Note: this code is copied on purpose from default codec as we want to call "this" `Send` and `Expect`
- if err := ctx.Err(); err != nil {
- return errors.Wrap(err, "Not sending message as context is aborted")
- }
- log.Trace().Msg("Sending request")
- // Send the actual message
- err := t.Send(message)
- if err != nil {
- return errors.Wrap(err, "Error sending the request")
- }
- return t.Expect(ctx, acceptsMessage, handleMessage, handleError, ttl)
-}
-
-func (t *TransactionStateMachine) handleOutboundMessage(message spi.Message) (handled bool, err error) {
- switch message := message.(type) {
- case readWriteModel.BVLCExactly:
- bvlc := message
- var npdu readWriteModel.NPDU
- if npduRetriever, ok := bvlc.(interface{ GetNpdu() readWriteModel.NPDU }); ok {
- npdu = npduRetriever.GetNpdu()
- } else {
- log.Debug().Msgf("bvlc has no way to give a npdu %T", bvlc)
- return false, nil
- }
- if npdu.GetControl().GetMessageTypeFieldPresent() {
- log.Trace().Msg("Message type field present")
- return false, nil
- }
- var entryForDestination = DeviceEntryDefault
- if npdu.GetControl().GetDestinationSpecified() {
- if retrievedEntry, err := t.deviceInventory.getEntryForDestination(npdu.GetDestinationAddress()); err != nil {
- // Get information from the device first
- // TODO: get information with who-has maybe or directed... not sure now
- // TODO: set entry once received
- _ = retrievedEntry
- }
- }
- // TODO: should we continue if we don't have a destination
- _ = entryForDestination
- apdu := npdu.GetApdu()
- switch apdu := apdu.(type) {
- case readWriteModel.APDUConfirmedRequestExactly:
- // TODO: this is a "client" request
- // TODO: check if adpu length is the magic number (it should be "unencoded")
- return false, nil
- case readWriteModel.APDUComplexAckExactly:
- // TODO: this is a "server" response
- // TODO: check if adpu length is the magic number (it should be "unencoded")
- return false, nil
- default:
- log.Trace().Msgf("APDU type not relevant %T present", apdu)
- return false, nil
- }
- default:
- log.Trace().Msgf("Message type not relevant %T present", message)
- return false, nil
- }
-}
-
-// TODO: this is a placeholder for a tasking framework
-type _Task struct {
- taskTime time.Time
- isScheduled bool
-}
-
-func (t *_Task) installTask(when *time.Time, delta *time.Duration) {
- // TODO: schedule task
-}
-
-func (t *_Task) suspendTask() {
- // TODO: suspend task
-}
-
-func (t *_Task) resume() {
- // TODO: resume task
-}
-
-type OneShotTask struct {
- _Task
-}
-
-// TODO: this is the interface to the outside for the SSM // TODO: maybe we should port that as non interface first
-type ServiceAccessPoint interface {
- GetDeviceInventory() *DeviceInventory
- GetLocalDevice() DeviceEntry
- GetProposedWindowSize() uint8
- Request(apdu readWriteModel.APDU)
- SapRequest(apdu readWriteModel.APDU)
- SapResponse(apdu readWriteModel.APDU)
- // TODO: wrap that properly
- GetClientTransactions() []interface{}
- // TODO: wrap that properly
- GetServerTransactions() []interface{}
- GetApplicationTimeout() *uint
-}
-
-// TODO: interface to client // TODO: maybe we should port that as non interface first
-type Client interface {
- request(apdu readWriteModel.APDU)
- confirmation(apdu readWriteModel.APDU)
-}
-
type SSMState uint8
const (
@@ -229,11 +72,22 @@ type segmentAPDU struct {
isAck bool
}
+type SSMSAPRequirements interface {
+ _ServiceAccessPoint
+ _Client
+ GetDeviceInventory() *DeviceInventory
+ GetLocalDevice() DeviceEntry
+ GetProposedWindowSize() uint8
+ GetClientTransactions() []*ClientSSM
+ GetServerTransactions() []*ServerSSM
+ GetApplicationTimeout() uint
+}
+
// SSM - Segmentation State Machine
type SSM struct {
OneShotTask
- ssmSAP ServiceAccessPoint
+ ssmSAP SSMSAPRequirements
pduAddress []byte
deviceEntry *DeviceEntry
@@ -260,7 +114,7 @@ type SSM struct {
maxApduLengthAccepted *readWriteModel.MaxApduLengthAccepted
}
-func NewSSM(sap ServiceAccessPoint, pduAddress []byte) (SSM, error) {
+func NewSSM(sap SSMSAPRequirements, pduAddress []byte) (SSM, error) {
log.Debug().Interface("sap", sap).Bytes("pdu_address", pduAddress).Msg("init")
deviceEntry, err := sap.GetDeviceInventory().getEntryForDestination(pduAddress)
if err != nil {
@@ -330,14 +184,14 @@ func (s *SSM) setSegmentationContext(apdu readWriteModel.APDU) error {
if apdu.GetSegmentedMessage() || apdu.GetMoreFollows() {
return errors.New("Can't handle already segmented message")
}
- bytes, err := apdu.GetServiceRequest().Serialize()
+ serializedBytes, err := apdu.GetServiceRequest().Serialize()
if err != nil {
return errors.Wrap(err, "Can serialize service request")
}
segmentAPDU := segmentAPDU{
originalApdu: apdu,
originalInvokeId: apdu.GetInvokeId(),
- serviceBytes: bytes,
+ serviceBytes: serializedBytes,
serviceChoice: apdu.GetServiceRequest().GetServiceChoice(),
}
s.segmentAPDU = &segmentAPDU
@@ -345,13 +199,13 @@ func (s *SSM) setSegmentationContext(apdu readWriteModel.APDU) error {
if apdu.GetSegmentedMessage() || apdu.GetMoreFollows() {
return errors.New("Can't handle already segmented message")
}
- bytes, err := apdu.GetServiceAck().Serialize()
+ serializedBytes, err := apdu.GetServiceAck().Serialize()
if err != nil {
return errors.Wrap(err, "Can serialize service request")
}
segmentAPDU := segmentAPDU{
originalApdu: apdu,
- serviceBytes: bytes,
+ serviceBytes: serializedBytes,
serviceChoice: apdu.GetServiceAck().GetServiceChoice(),
isAck: true,
}
@@ -435,20 +289,20 @@ func (s *SSM) appendSegment(apdu readWriteModel.APDU) error {
if apdu.GetSegmentedMessage() || apdu.GetMoreFollows() {
return errors.New("Can't handle already segmented message")
}
- bytes, err := apdu.GetServiceRequest().Serialize()
+ serializedBytes, err := apdu.GetServiceRequest().Serialize()
if err != nil {
return errors.Wrap(err, "Can serialize service request")
}
- s.segmentAPDU.serviceBytes = append(s.segmentAPDU.serviceBytes, bytes...)
+ s.segmentAPDU.serviceBytes = append(s.segmentAPDU.serviceBytes, serializedBytes...)
case readWriteModel.APDUComplexAckExactly:
if apdu.GetSegmentedMessage() || apdu.GetMoreFollows() {
return errors.New("Can't handle already segmented message")
}
- bytes, err := apdu.GetServiceAck().Serialize()
+ serializedBytes, err := apdu.GetServiceAck().Serialize()
if err != nil {
return errors.Wrap(err, "Can serialize service request")
}
- s.segmentAPDU.serviceBytes = append(s.segmentAPDU.serviceBytes, bytes...)
+ s.segmentAPDU.serviceBytes = append(s.segmentAPDU.serviceBytes, serializedBytes...)
default:
return errors.Errorf("invalid APDU type %T", apdu)
}
@@ -467,7 +321,9 @@ func (s *SSM) fillWindow(sequenceNumber uint8) error {
if err != nil {
return errors.Wrapf(err, "Error sending out segment %d", i)
}
- s.ssmSAP.Request(apdu)
+ if err := s.ssmSAP.Request(apdu); err != nil {
+ log.Debug().Err(err).Msg("error sending request")
+ }
if moreFollows {
s.sentAllSegments = true
}
@@ -479,7 +335,7 @@ type ClientSSM struct {
SSM
}
-func NewClientSSM(sap ServiceAccessPoint, pduAddress []byte) (*ClientSSM, error) {
+func NewClientSSM(sap SSMSAPRequirements, pduAddress []byte) (*ClientSSM, error) {
log.Debug().Interface("sap", sap).Bytes("pduAddress", pduAddress).Msg("init")
ssm, err := NewSSM(sap, pduAddress)
if err != nil {
@@ -514,18 +370,17 @@ func (s *ClientSSM) setState(newState SSMState, timer *uint) error {
return nil
}
-// request This function is called by client transaction functions when it wants to send a message to the device
-func (s *ClientSSM) request(apdu readWriteModel.APDU) {
+// Request This function is called by client transaction functions when it wants to send a message to the device
+func (s *ClientSSM) Request(apdu readWriteModel.APDU) 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
- s.ssmSAP.Request(apdu)
+ return s.ssmSAP.Request(apdu)
}
-// TODO: maybe use another name for that
-// indication This function is called after the device has bound a new transaction and wants to start the process
+// 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 readWriteModel.APDU) error {
+func (s *ClientSSM) Indication(apdu readWriteModel.APDU) error { // TODO: maybe use another name for that
log.Debug().Msgf("indication\n%s", apdu)
// make sure we're getting confirmed requests
var apduConfirmedRequest readWriteModel.APDUConfirmedRequest
@@ -568,8 +423,7 @@ func (s *ClientSSM) indication(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "Error creating abort")
}
- s.response(abort)
- return nil
+ return s.Response(abort)
}
if s.deviceEntry == nil {
@@ -580,8 +434,7 @@ func (s *ClientSSM) indication(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "Error creating abort")
}
- s.response(abort)
- return nil
+ return s.Response(abort)
}
// make sure we don't exceed the number of segments in our request that the server said it was willing to accept
@@ -595,8 +448,7 @@ func (s *ClientSSM) indication(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "Error creating abort")
}
- s.response(abort)
- return nil
+ return s.Response(abort)
}
}
@@ -625,23 +477,22 @@ func (s *ClientSSM) indication(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "error getting segment")
}
- s.request(segment)
- return nil
+ return s.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 readWriteModel.APDU) {
+// Response This function is called by client transaction functions when they want to send a message to the application.
+func (s *ClientSSM) Response(apdu readWriteModel.APDU) error {
log.Debug().Msgf("response\n%s", apdu)
// make sure it has a good source and destination
// TODO: check if source == s.pduAddress
// TODO: check if
// send it to the application
- s.ssmSAP.SapResponse(apdu)
+ return s.ssmSAP.SapResponse(apdu)
}
-// confirmation This function is called by the device for all upstream messages related to the transaction.
-func (s *ClientSSM) confirmation(apdu readWriteModel.APDU) error {
+// Confirmation This function is called by the device for all upstream messages related to the transaction.
+func (s *ClientSSM) Confirmation(apdu readWriteModel.APDU) error {
log.Debug().Msgf("confirmation\n%s", apdu)
switch s.state {
@@ -728,8 +579,12 @@ func (s *ClientSSM) segmentedRequest(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- s.request(abort) // send it ot the device
- s.response(abort) // send it ot the application
+ if err := s.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
+ log.Debug().Err(err).Msg("error sending response")
+ }
} else {
if err := s.setState(COMPLETED, nil); err != nil {
return errors.Wrap(err, "error switching state")
@@ -743,14 +598,20 @@ func (s *ClientSSM) segmentedRequest(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- s.request(abort) // send it ot the device
- s.response(abort) // send it ot the application
+ if err := s.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
+ log.Debug().Err(err).Msg("error sending response")
+ }
} else if !apdu.GetSegmentedMessage() {
// ack is not segmented
if err := s.setState(COMPLETED, nil); err != nil {
return errors.Wrap(err, "error switching state")
}
- s.response(apdu)
+ if err := s.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 {
@@ -771,7 +632,9 @@ func (s *ClientSSM) segmentedRequest(apdu readWriteModel.APDU) error {
if err := s.setState(COMPLETED, nil); err != nil {
return errors.Wrap(err, "error switching state")
}
- s.response(apdu)
+ if err := s.Response(apdu); err != nil {
+ log.Debug().Err(err).Msg("error sending response")
+ }
default:
return errors.Errorf("Invalid apdu %T", apdu)
}
@@ -792,7 +655,9 @@ func (s *ClientSSM) segmentedRequestTimeout() error {
if err != nil {
return errors.Wrap(err, "error getting first segment")
}
- s.request(apdu)
+ if err := s.Request(apdu); err != nil {
+ log.Debug().Err(err).Msg("error sending request")
+ }
} else {
if err := s.fillWindow(s.initialSequenceNumber); err != nil {
return errors.Wrap(err, "error filling window")
@@ -805,7 +670,9 @@ func (s *ClientSSM) segmentedRequestTimeout() error {
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- s.response(abort)
+ if err := s.Response(abort); err != nil {
+ log.Debug().Err(err).Msg("error sending response")
+ }
}
return nil
}
@@ -820,14 +687,18 @@ func (s *ClientSSM) awaitConfirmation(apdu readWriteModel.APDU) error {
if err := s.setState(ABORTED, nil); err != nil {
return errors.Wrap(err, "error switching state")
}
- s.response(apdu)
+ if err := s.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(COMPLETED, nil); err != nil {
return errors.Wrap(err, "error switching state")
}
- s.response(apdu)
+ if err := s.Response(apdu); err != nil {
+ log.Debug().Err(err).Msg("error sending response")
+ }
case readWriteModel.APDUComplexAckExactly:
log.Debug().Msg("complex ack")
@@ -837,7 +708,9 @@ func (s *ClientSSM) awaitConfirmation(apdu readWriteModel.APDU) error {
if err := s.setState(COMPLETED, nil); err != nil {
return errors.Wrap(err, "error switching state")
}
- s.response(apdu)
+ if err := s.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 {
log.Debug().Msg("local device can't receive segmented messages")
@@ -845,7 +718,9 @@ func (s *ClientSSM) awaitConfirmation(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- s.response(abort)
+ if err := s.Response(abort); err != nil {
+ log.Debug().Err(err).Msg("error sending response")
+ }
} else if *apdu.GetSequenceNumber() == 0 {
log.Debug().Msg("segmented response")
@@ -863,7 +738,9 @@ func (s *ClientSSM) awaitConfirmation(apdu readWriteModel.APDU) error {
// send back a segment ack
segmentAck := readWriteModel.NewAPDUSegmentAck(false, false, s.invokeId, s.initialSequenceNumber, *s.actualWindowSize, 0)
- s.request(segmentAck)
+ if err := s.Request(segmentAck); err != nil {
+ log.Debug().Err(err).Msg("error sending request")
+ }
} else {
log.Debug().Msg("Invalid apdu in this state")
@@ -871,8 +748,12 @@ func (s *ClientSSM) awaitConfirmation(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- s.request(abort) // send it to the device
- s.response(abort) // send it to the application
+ if err := s.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
+ log.Debug().Err(err).Msg("error sending response")
+ }
}
case readWriteModel.APDUSegmentAckExactly:
log.Debug().Msg("segment ack(!?)")
@@ -893,7 +774,7 @@ func (s *ClientSSM) awaitConfirmationTimeout() error {
// 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
+ if err := s.Indication(s.segmentAPDU.originalApdu); err != nil { // TODO: check that it is really the intention to re-send the original apdu here
return err
}
s.retryCount = saveCount
@@ -904,7 +785,9 @@ func (s *ClientSSM) awaitConfirmationTimeout() error {
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- s.response(abort)
+ if err := s.Response(abort); err != nil {
+ log.Debug().Err(err).Msg("error sending response")
+ }
}
return nil
}
@@ -921,8 +804,12 @@ func (s *ClientSSM) segmentedConfirmation(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- s.request(abort) // send it to the device
- s.response(abort) // send it to the application
+ if err := s.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
+ log.Debug().Err(err).Msg("error sending response")
+ }
}
// it must be segmented
@@ -931,8 +818,12 @@ func (s *ClientSSM) segmentedConfirmation(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- s.request(abort) // send it to the device
- s.response(abort) // send it to the application
+ if err := s.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
+ log.Debug().Err(err).Msg("error sending response")
+ }
}
// proper segment number
@@ -942,7 +833,9 @@ func (s *ClientSSM) segmentedConfirmation(apdu readWriteModel.APDU) error {
// segment received out of order
s.restartTimer(s.segmentTimeout)
segmentAck := readWriteModel.NewAPDUSegmentAck(true, false, s.invokeId, s.initialSequenceNumber, *s.actualWindowSize, 0)
- s.request(segmentAck)
+ if err := s.Request(segmentAck); err != nil {
+ log.Debug().Err(err).Msg("error sending request")
+ }
return nil
}
@@ -960,7 +853,9 @@ func (s *ClientSSM) segmentedConfirmation(apdu readWriteModel.APDU) error {
// send final ack
segmentAck := readWriteModel.NewAPDUSegmentAck(false, false, s.invokeId, s.lastSequenceNumber, *s.actualWindowSize, 0)
- s.request(segmentAck)
+ if err := s.Request(segmentAck); err != nil {
+ log.Debug().Err(err).Msg("error sending request")
+ }
if err := s.setState(COMPLETED, nil); err != nil {
return errors.Wrap(err, "error switching state")
@@ -971,14 +866,18 @@ func (s *ClientSSM) segmentedConfirmation(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "error parsing apdu")
}
- s.response(parse)
+ if err := s.Response(parse); err != nil {
+ log.Debug().Err(err).Msg("error sending response")
+ }
} else if *apduComplexAck.GetSequenceNumber() == s.initialSequenceNumber+*s.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)
- s.request(segmentAck)
+ if err := s.Request(segmentAck); err != nil { // send it ot the device
+ log.Debug().Err(err).Msg("error sending request")
+ }
} else {
log.Debug().Msg("Wait for more segments")
@@ -995,8 +894,7 @@ func (s *ClientSSM) segmentedConfirmationTimeout() error {
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- s.response(abort)
- return nil
+ return s.Response(abort)
}
type ServerSSM struct {
@@ -1004,7 +902,7 @@ type ServerSSM struct {
segmentedResponseAccepted bool
}
-func NewServerSSM(sap ServiceAccessPoint, pduAddress []byte) (*ServerSSM, error) {
+func NewServerSSM(sap SSMSAPRequirements, pduAddress []byte) (*ServerSSM, error) {
log.Debug().Interface("sap", sap).Bytes("pduAddress", pduAddress).Msg("init")
ssm, err := NewSSM(sap, pduAddress)
if err != nil {
@@ -1040,18 +938,17 @@ func (s *ServerSSM) setState(newState SSMState, timer *uint) error {
return nil
}
-// request This function is called by transaction functions to send to the application
-func (s *ServerSSM) request(apdu readWriteModel.APDU) {
+// Request This function is called by transaction functions to send to the application
+func (s *ServerSSM) Request(apdu readWriteModel.APDU) 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
- s.ssmSAP.SapRequest(apdu)
+ return s.ssmSAP.SapRequest(apdu)
}
-// TODO: maybe use another name for that
-// indication This function is called for each downstream packet related to
+// Indication This function is called for each downstream packet related to
// the transaction
-func (s *ServerSSM) indication(apdu readWriteModel.APDU) error {
+func (s *ServerSSM) Indication(apdu readWriteModel.APDU) error { // TODO: maybe use another name for that
log.Debug().Msgf("indication\n%s", apdu)
// make sure we're getting confirmed requests
@@ -1069,20 +966,20 @@ func (s *ServerSSM) indication(apdu readWriteModel.APDU) error {
}
}
-// response This function is called by client transaction functions when they want to send a message to the application.
-func (s *ServerSSM) response(apdu readWriteModel.APDU) {
+// Response This function is called by client transaction functions when they want to send a message to the application.
+func (s *ServerSSM) Response(apdu readWriteModel.APDU) error {
log.Debug().Msgf("response\n%s", apdu)
// make sure it has a good source and destination
// TODO: check if source == none
// TODO: check if destnation = s.pduAddress
// send it via the device
- s.ssmSAP.Request(apdu)
+ return s.ssmSAP.Request(apdu)
}
-// confirmation This function is called when the application has provided a response and needs it to be sent to the
+// 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 readWriteModel.APDU) error {
+func (s *ServerSSM) Confirmation(apdu readWriteModel.APDU) error {
log.Debug().Msgf("confirmation\n%s", apdu)
// check to see we are in the correct state
@@ -1100,8 +997,7 @@ func (s *ServerSSM) confirmation(apdu readWriteModel.APDU) error {
}
// end the response to the device
- s.response(apdu)
- return nil
+ return s.Response(apdu)
// simple response
case readWriteModel.APDUSimpleAckExactly, readWriteModel.APDUErrorExactly, readWriteModel.APDURejectExactly:
log.Debug().Msg("simple ack, error or reject")
@@ -1112,8 +1008,7 @@ func (s *ServerSSM) confirmation(apdu readWriteModel.APDU) error {
}
// send the response to the device
- s.response(apdu)
- return nil
+ return s.Response(apdu)
// complex ack
case readWriteModel.APDUComplexAckExactly:
log.Debug().Msg("complex ack")
@@ -1155,8 +1050,7 @@ func (s *ServerSSM) confirmation(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "Error creating abort")
}
- s.response(abort)
- return nil
+ return s.Response(abort)
}
// make sure client supports segmented receive
@@ -1166,8 +1060,7 @@ func (s *ServerSSM) confirmation(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "Error creating abort")
}
- s.response(abort)
- return nil
+ return s.Response(abort)
}
// make sure we don't exceed the number of segments in our response that the client said it was willing to accept
@@ -1178,8 +1071,7 @@ func (s *ServerSSM) confirmation(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "Error creating abort")
}
- s.response(abort)
- return nil
+ return s.Response(abort)
}
}
@@ -1190,7 +1082,9 @@ func (s *ServerSSM) confirmation(apdu readWriteModel.APDU) error {
// send out the first segment (or the whole thing)
if s.segmentCount == 1 {
- s.response(apdu)
+ if err := s.Response(apdu); err != nil {
+ log.Debug().Err(err).Msg("error sending response")
+ }
if err := s.setState(COMPLETED, nil); err != nil {
return errors.Wrap(err, "Error setting state to aborted")
}
@@ -1199,7 +1093,9 @@ func (s *ServerSSM) confirmation(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "error getting first segment")
}
- s.response(segment)
+ if err := s.Response(segment); err != nil {
+ log.Debug().Err(err).Msg("error sending response")
+ }
if err := s.setState(SEGMENTED_RESPONSE, nil); err != nil {
return errors.Wrap(err, "Error setting state to aborted")
}
@@ -1304,8 +1200,7 @@ func (s *ServerSSM) idle(apdu readWriteModel.APDU) error {
if err := s.setState(AWAIT_RESPONSE, nil); err != nil {
return errors.Wrap(err, "Error setting state to aborted")
}
- s.request(apdu)
- return nil
+ return s.Request(apdu)
}
// make sure we support segmented requests
@@ -1314,8 +1209,7 @@ func (s *ServerSSM) idle(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- s.response(abort)
- return nil
+ return s.Response(abort)
}
// save the response and set the segmentation context
@@ -1338,8 +1232,7 @@ func (s *ServerSSM) idle(apdu readWriteModel.APDU) error {
// send back a segment ack
segack := readWriteModel.NewAPDUSegmentAck(false, true, s.invokeId, s.initialSequenceNumber, *s.actualWindowSize, 0)
log.Debug().Msgf("segAck: %s", segack)
- s.response(segack)
- return nil
+ return s.Response(segack)
}
func (s *ServerSSM) segmentedRequest(apdu readWriteModel.APDU) error {
@@ -1350,8 +1243,7 @@ func (s *ServerSSM) segmentedRequest(apdu readWriteModel.APDU) error {
if err := s.setState(COMPLETED, nil); err != nil {
return errors.Wrap(err, "Error setting state to aborted")
}
- s.response(apdu)
- return nil
+ return s.Response(apdu)
}
// the only messages we should be getting are confirmed requests
@@ -1361,8 +1253,12 @@ func (s *ServerSSM) segmentedRequest(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- s.request(abort) // send it ot the device
- s.response(abort) // send it ot the application
+ if err := s.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
+ log.Debug().Err(err).Msg("error sending response")
+ }
} else {
apduConfirmedRequest = castedApdu
}
@@ -1373,8 +1269,12 @@ func (s *ServerSSM) segmentedRequest(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- s.request(abort) // send it ot the device
- s.response(abort) // send it ot the application
+ if err := s.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
+ log.Debug().Err(err).Msg("error sending response")
+ }
}
// proper segment number
@@ -1386,8 +1286,7 @@ func (s *ServerSSM) segmentedRequest(apdu readWriteModel.APDU) error {
// send back a segment ack
segack := readWriteModel.NewAPDUSegmentAck(true, true, s.invokeId, s.initialSequenceNumber, *s.actualWindowSize, 0)
- s.response(segack)
- return nil
+ return s.Response(segack)
}
// add the data
@@ -1404,10 +1303,13 @@ func (s *ServerSSM) segmentedRequest(apdu readWriteModel.APDU) error {
// send back the final segment ack
segack := readWriteModel.NewAPDUSegmentAck(false, true, s.invokeId, s.lastSequenceNumber, *s.actualWindowSize, 0)
- s.response(segack)
+ if err := s.Response(segack); err != nil {
+ log.Debug().Err(err).Msg("error sending response")
+ }
// forward the whole thing to the application
- if err := s.setState(AWAIT_RESPONSE, s.ssmSAP.GetApplicationTimeout()); err != nil {
+ applicationTimeout := s.ssmSAP.GetApplicationTimeout()
+ if err := s.setState(AWAIT_RESPONSE, &applicationTimeout); err != nil {
return errors.Wrap(err, "Error setting state to aborted")
}
// TODO: here we need to rebuild again yada yada
@@ -1417,7 +1319,9 @@ func (s *ServerSSM) segmentedRequest(apdu readWriteModel.APDU) error {
if err != nil {
return errors.Wrap(err, "error parsing apdu")
}
- s.request(parse)
+ if err := s.Request(parse); err != nil {
+ log.Debug().Err(err).Msg("error sending request")
+ }
} else if *apduConfirmedRequest.GetSequenceNumber() == s.initialSequenceNumber+*s.actualWindowSize {
log.Debug().Msg("last segment in the group")
@@ -1426,7 +1330,9 @@ func (s *ServerSSM) segmentedRequest(apdu readWriteModel.APDU) error {
// send back a segment ack
segack := readWriteModel.NewAPDUSegmentAck(false, true, s.invokeId, s.initialSequenceNumber, *s.actualWindowSize, 0)
- s.response(segack)
+ if err := s.Response(segack); err != nil {
+ log.Debug().Err(err).Msg("error sending response")
+ }
} else {
// wait for more segments
s.restartTimer(s.segmentTimeout)
@@ -1458,7 +1364,9 @@ func (s *ServerSSM) awaitResponse(apdu readWriteModel.APDU) error {
if err := s.setState(ABORTED, nil); err != nil {
return errors.Wrap(err, "Error setting state to aborted")
}
- s.request(apdu)
+ if err := s.Request(apdu); err != nil { // send it ot the device
+ log.Debug().Err(err).Msg("error sending request")
+ }
default:
return errors.Errorf("invalid APDU %T", apdu)
}
@@ -1474,7 +1382,9 @@ func (s *ServerSSM) awaitResponseTimeout() error {
if err != nil {
return errors.Wrap(err, "error creating abort")
}
- s.request(abort)
+ if err := s.Request(abort); err != nil {
+ log.Debug().Err(err).Msg("error sending request")
+ }
return nil
}
@@ -1517,7 +1427,9 @@ func (s *ServerSSM) segmentedResponse(apdu readWriteModel.APDU) error {
if err := s.setState(COMPLETED, nil); err != nil {
return errors.Wrap(err, "Error setting state to aborted")
}
- s.response(apdu)
+ if err := s.Response(apdu); err != nil { // send it ot the application
+ log.Debug().Err(err).Msg("error sending response")
+ }
default:
return errors.Errorf("Invalid APDU %T", apdu)
}
@@ -1544,8 +1456,8 @@ func (s *ServerSSM) segmentedResponseTimeout() error {
}
type StateMachineAccessPoint struct {
- Client
- ServiceAccessPoint
+ *Client
+ *ServiceAccessPoint
localDevice DeviceEntry
deviceInventory *DeviceInventory
@@ -1558,18 +1470,15 @@ type StateMachineAccessPoint struct {
segmentationSupported readWriteModel.BACnetSegmentation
segmentTimeout int
maxSegmentsAccepted int
- proposedWindowSize int
+ proposedWindowSize uint8
dccEnableDisable readWriteModel.BACnetConfirmedServiceRequestDeviceCommunicationControlEnableDisable
- applicationTimeout int
+ applicationTimeout uint
}
-func NewStateMachineAccessPoint(localDevice DeviceEntry, deviceInventory *DeviceInventory, sapID int, cid int) StateMachineAccessPoint {
+func NewStateMachineAccessPoint(localDevice DeviceEntry, deviceInventory *DeviceInventory, sapID *int, cid *int) (*StateMachineAccessPoint, error) {
log.Debug().Msgf("NewStateMachineAccessPoint localDevice=%v deviceInventory=%v sap=%v cid=%v", localDevice, deviceInventory, sapID, cid)
- // basic initialization
- // TODO: init client
- // TODO: init sap
- return StateMachineAccessPoint{
+ s := &StateMachineAccessPoint{
// save a reference to the device information cache
localDevice: localDevice,
deviceInventory: deviceInventory,
@@ -1599,6 +1508,18 @@ func NewStateMachineAccessPoint(localDevice DeviceEntry, deviceInventory *Device
// layer to form a response and send it
applicationTimeout: 3000,
}
+ // basic initialization
+ client, err := NewClient(cid, s)
+ if err != nil {
+ return nil, errors.Wrapf(err, "error building client for %d", cid)
+ }
+ s.Client = client
+ serviceAccessPoint, err := NewServiceAccessPoint(sapID, s)
+ if err != nil {
+ return nil, errors.Wrapf(err, "error building serviceAccessPoint for %d", sapID)
+ }
+ s.ServiceAccessPoint = serviceAccessPoint
+ return s, nil
}
// getNextInvokeId Called by clients to get an unused invoke ID
@@ -1628,8 +1549,8 @@ func (s *StateMachineAccessPoint) getNextInvokeId(address []byte) (uint8, error)
}
}
-// confirmation Packets coming up the stack are APDU's
-func (s *StateMachineAccessPoint) confirmation(apdu readWriteModel.APDU, pduSource []byte) error {
+// 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
log.Debug().Msgf("confirmation\n%s", apdu)
// check device communication control
@@ -1676,12 +1597,14 @@ func (s *StateMachineAccessPoint) confirmation(apdu readWriteModel.APDU, pduSour
}
// let it run with the apdu
- if err := tr.indication(apdu); err != nil {
+ if err := tr.Indication(apdu); err != nil {
return errors.Wrap(err, "error runnning indication")
}
case readWriteModel.APDUUnconfirmedRequestExactly:
// deliver directly to the application
- s.SapRequest(apdu)
+ if err := s.SapRequest(apdu); err != nil {
+ log.Debug().Err(err).Msg("error sending request")
+ }
case readWriteModel.APDUSimpleAckExactly, readWriteModel.APDUComplexAckExactly, readWriteModel.APDUErrorExactly, readWriteModel.APDURejectExactly:
// find the client transaction this is acking
var tr *ClientSSM
@@ -1696,7 +1619,7 @@ func (s *StateMachineAccessPoint) confirmation(apdu readWriteModel.APDU, pduSour
}
// send the packet on to the transaction
- if err := tr.confirmation(apdu); err != nil {
+ if err := tr.Confirmation(apdu); err != nil {
return errors.Wrap(err, "error running confirmation")
}
case readWriteModel.APDUAbortExactly:
@@ -1714,7 +1637,7 @@ func (s *StateMachineAccessPoint) confirmation(apdu readWriteModel.APDU, pduSour
}
// send the packet on to the transaction
- if err := tr.confirmation(apdu); err != nil {
+ if err := tr.Confirmation(apdu); err != nil {
return errors.Wrap(err, "error running confirmation")
}
} else {
@@ -1731,7 +1654,7 @@ func (s *StateMachineAccessPoint) confirmation(apdu readWriteModel.APDU, pduSour
}
// send the packet on to the transaction
- if err := tr.indication(apdu); err != nil {
+ if err := tr.Indication(apdu); err != nil {
return errors.Wrap(err, "error running indication")
}
}
@@ -1750,7 +1673,7 @@ func (s *StateMachineAccessPoint) confirmation(apdu readWriteModel.APDU, pduSour
}
// send the packet on to the transaction
- if err := tr.confirmation(apdu); err != nil {
+ if err := tr.Confirmation(apdu); err != nil {
return errors.Wrap(err, "error running confirmation")
}
} else {
@@ -1767,7 +1690,7 @@ func (s *StateMachineAccessPoint) confirmation(apdu readWriteModel.APDU, pduSour
}
// send the packet on to the transaction
- if err := tr.indication(apdu); err != nil {
+ if err := tr.Indication(apdu); err != nil {
return errors.Wrap(err, "error running indication")
}
}
@@ -1777,8 +1700,8 @@ func (s *StateMachineAccessPoint) confirmation(apdu readWriteModel.APDU, pduSour
return nil
}
-// sapIndication This function is called when the application is requesting a new transaction as a client.
-func (s *StateMachineAccessPoint) sapIndication(apdu readWriteModel.APDU, pduDestination []byte) error {
+// SapIndication This function is called when the application is requesting a new transaction as a client.
+func (s *StateMachineAccessPoint) SapIndication(apdu readWriteModel.APDU, pduDestination []byte) error {
log.Debug().Msgf("sapIndication\n%s", apdu)
// check device communication control
@@ -1801,7 +1724,9 @@ func (s *StateMachineAccessPoint) sapIndication(apdu readWriteModel.APDU, pduDes
switch apdu := apdu.(type) {
case readWriteModel.APDUUnconfirmedRequestExactly:
// deliver to the device
- s.request(apdu)
+ if err := s.Request(apdu); err != nil {
+ log.Debug().Err(err).Msg("error sending the request")
+ }
case readWriteModel.APDUConfirmedRequestExactly:
// make sure it has an invoke ID
// 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)
@@ -1825,7 +1750,7 @@ func (s *StateMachineAccessPoint) sapIndication(apdu readWriteModel.APDU, pduDes
s.clientTransactions = append(s.clientTransactions, tr)
// let it run
- if err := tr.indication(apdu); err != nil {
+ if err := tr.Indication(apdu); err != nil {
return errors.Wrap(err, "error doing indication")
}
default:
@@ -1835,9 +1760,9 @@ func (s *StateMachineAccessPoint) sapIndication(apdu readWriteModel.APDU, pduDes
return nil
}
-// sapConfirmation This function is called when the application is responding to a request, the apdu may be a simple
+// 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 readWriteModel.APDU, pduDestination []byte) error {
+func (s *StateMachineAccessPoint) SapConfirmation(apdu readWriteModel.APDU, pduDestination []byte) error {
log.Debug().Msgf("sapConfirmation\n%s", apdu)
switch apdu.(type) {
case readWriteModel.APDUSimpleAckExactly, readWriteModel.APDUComplexAckExactly, readWriteModel.APDUErrorExactly, readWriteModel.APDURejectExactly:
@@ -1854,7 +1779,7 @@ func (s *StateMachineAccessPoint) sapConfirmation(apdu readWriteModel.APDU, pduD
}
// pass control to the transaction
- if err := tr.confirmation(apdu); err != nil {
+ if err := tr.Confirmation(apdu); err != nil {
return errors.Wrap(err, "error running confirmation")
}
default:
@@ -1862,3 +1787,89 @@ func (s *StateMachineAccessPoint) sapConfirmation(apdu readWriteModel.APDU, pduD
}
return nil
}
+
+func (s *StateMachineAccessPoint) GetDeviceInventory() *DeviceInventory {
+ return s.deviceInventory
+}
+
+func (s *StateMachineAccessPoint) GetLocalDevice() DeviceEntry {
+ return s.localDevice
+}
+
+func (s *StateMachineAccessPoint) GetProposedWindowSize() uint8 {
+ return s.proposedWindowSize
+}
+
+func (s *StateMachineAccessPoint) GetClientTransactions() []*ClientSSM {
+ return s.clientTransactions
+}
+
+func (s *StateMachineAccessPoint) GetServerTransactions() []*ServerSSM {
+ return s.serverTransactions
+}
+
+func (s *StateMachineAccessPoint) GetApplicationTimeout() uint {
+ return s.applicationTimeout
+}
+
+type ApplicationServiceAccessPoint struct {
+ *ApplicationServiceElement
+ *ServiceAccessPoint
+}
+
+func NewApplicationServiceAccessPoint(aseID *int, sapID *int) (*ApplicationServiceAccessPoint, error) {
+ a := &ApplicationServiceAccessPoint{}
+ applicationServiceElement, err := NewApplicationServiceElement(aseID, a)
+ if err != nil {
+ return nil, errors.Wrap(err, "error creating application service element")
+ }
+ a.ApplicationServiceElement = applicationServiceElement
+ serviceAccessPoint, err := NewServiceAccessPoint(sapID, a)
+ if err != nil {
+ return nil, errors.Wrap(err, "error creating service access point")
+ }
+ a.ServiceAccessPoint = serviceAccessPoint
+ return a, nil
+}
+
+// TODO: big WIP
+func (a *ApplicationServiceAccessPoint) Indication(apdu readWriteModel.APDU) error {
+ log.Debug().Msgf("indication\n%s", apdu)
+
+ switch apdu := apdu.(type) {
+ case readWriteModel.APDUConfirmedRequestExactly:
+ //assume no errors found
+ var errorFound error
+ if !readWriteModel.BACnetConfirmedServiceChoiceKnows(uint8(apdu.GetServiceRequest().GetServiceChoice())) {
+ errorFound = errors.New("unrecognized service")
+ }
+
+ if errorFound == nil {
+ errorFound = a.SapRequest(apdu)
+ }
+ // TODO: the handling here gets a bit different now... need to wrap the head around how to do this (error handling etc)
+
+ if errorFound == nil {
+ if err := a.SapResponse(apdu); err != nil {
+ return err
+ }
+ } else {
+ log.Debug().Err(errorFound).Msg("got error")
+ }
+
+ case readWriteModel.APDUUnconfirmedRequestExactly:
+ default:
+ return errors.Errorf("unknown PDU type %T", apdu)
+ }
+ return nil
+}
+
+// TODO: big WIP
+func (a *ApplicationServiceAccessPoint) Confirmation(apdu readWriteModel.APDU) error {
+ panic("not yet implemented")
+}
+
+// TODO: big WIP
+func (a *ApplicationServiceAccessPoint) SapConfirmation(apdu readWriteModel.APDU, pduDestination []byte) error {
+ panic("not yet implemented")
+}
diff --git a/plc4go/internal/bacnetip/CommunicationsModule.go b/plc4go/internal/bacnetip/CommunicationsModule.go
new file mode 100644
index 0000000000..f92cf24a8e
--- /dev/null
+++ b/plc4go/internal/bacnetip/CommunicationsModule.go
@@ -0,0 +1,391 @@
+/*
+ * 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 (
+ readWriteModel "github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model"
+ "github.com/pkg/errors"
+ "github.com/rs/zerolog/log"
+)
+
+// maps of named clients and servers
+var clientMap map[int]*Client
+var serverMap map[int]*Server
+
+// maps of named SAPs and ASEs
+var serviceMap map[int]*ServiceAccessPoint
+var elementMap map[int]*ApplicationServiceElement
+
+func init() {
+ clientMap = make(map[int]*Client)
+ serverMap = make(map[int]*Server)
+ serviceMap = make(map[int]*ServiceAccessPoint)
+ elementMap = make(map[int]*ApplicationServiceElement)
+}
+
+// _Client is an interface used for documentation
+type _Client interface {
+ Request(apdu readWriteModel.APDU) error
+ Confirmation(apdu readWriteModel.APDU) error
+ _setClientPeer(server _Server)
+}
+
+// Client is an "abstract" struct which is used in another struct as delegate
+type Client struct {
+ clientID *int
+ clientPeer _Server
+}
+
+func NewClient(cid *int, rootStruct _Client) (*Client, error) {
+ c := &Client{
+ clientID: cid,
+ }
+ if cid != nil {
+ if _, ok := clientMap[*cid]; ok {
+ return nil, errors.Errorf("already a client %d", *cid)
+ }
+ clientMap[*cid] = c
+
+ // automatically bind
+ if server, ok := serverMap[*cid]; ok {
+ if server.serverPeer != nil {
+ return nil, errors.Errorf("server %d already bound", *cid)
+ }
+
+ // Note: we need to pass the rootStruct (which should contain c as delegate) here
+ if err := bind(rootStruct, server); err != nil {
+ return nil, errors.Wrap(err, "error binding")
+ }
+ }
+ }
+ return c, nil
+}
+
+func (c *Client) Request(apdu readWriteModel.APDU) error {
+ log.Debug().Msgf("request\n%s", apdu)
+
+ if c.clientPeer == nil {
+ return errors.New("unbound client")
+ }
+ return c.clientPeer.Indication(apdu)
+}
+
+func (c *Client) Confirmation(readWriteModel.APDU) error {
+ panic("this should be implemented by outer struct")
+}
+
+func (c *Client) _setClientPeer(server _Server) {
+ c.clientPeer = server
+}
+
+// _Server is an interface used for documentation
+type _Server interface {
+ Indication(apdu readWriteModel.APDU) error
+ Response(apdu readWriteModel.APDU) error
+ _setServerPeer(serverPeer _Client)
+}
+
+// Server is an "abstract" struct which is used in another struct as delegate
+type Server struct {
+ serverID *int
+ serverPeer _Client
+}
+
+func NewServer(sid *int, rootStruct _Server) (*Server, error) {
+ s := &Server{
+ serverID: sid,
+ }
+ if sid != nil {
+ if _, ok := serverMap[*sid]; ok {
+ return nil, errors.Errorf("already a server %d", *sid)
+ }
+ serverMap[*sid] = s
+
+ // automatically bind
+ if client, ok := clientMap[*sid]; ok {
+ if client.clientPeer != nil {
+ return nil, errors.Errorf("client %d already bound", *sid)
+ }
+
+ // Note: we need to pass the rootStruct (which should contain s as delegate) here
+ if err := bind(client, rootStruct); err != nil {
+ return nil, errors.Wrap(err, "error binding")
+ }
+ }
+ }
+ return s, nil
+}
+
+func (s *Server) Indication(readWriteModel.APDU) error {
+ panic("this should be implemented by outer struct")
+}
+
+func (s *Server) Response(apdu readWriteModel.APDU) error {
+ log.Debug().Msgf("response\n%s", apdu)
+
+ if s.serverPeer == nil {
+ return errors.New("unbound server")
+ }
+ return s.serverPeer.Confirmation(apdu)
+}
+
+func (s *Server) _setServerPeer(serverPeer _Client) {
+ s.serverPeer = serverPeer
+}
+
+// _ServiceAccessPoint is a interface used for documentation
+type _ServiceAccessPoint interface {
+ SapConfirmation(apdu readWriteModel.APDU, pduDestination []byte) error
+ SapRequest(apdu readWriteModel.APDU) error
+ SapIndication(apdu readWriteModel.APDU, pduDestination []byte) error
+ SapResponse(apdu readWriteModel.APDU) error
+ _setServiceElement(serviceElement _ApplicationServiceElement)
+}
+
+type ServiceAccessPoint struct {
+ serviceID *int
+ serviceElement _ApplicationServiceElement
+}
+
+func NewServiceAccessPoint(sapID *int, rootStruct _ServiceAccessPoint) (*ServiceAccessPoint, error) {
+ s := &ServiceAccessPoint{
+ serviceID: sapID,
+ }
+ if sapID != nil {
+ if _, ok := serviceMap[*sapID]; ok {
+ return nil, errors.Errorf("already a server %d", *sapID)
+ }
+ serviceMap[*sapID] = s
+
+ // automatically bind
+ if element, ok := elementMap[*sapID]; ok {
+ if element.elementService != nil {
+ return nil, errors.Errorf("application service element %d already bound", *sapID)
+ }
+
+ // Note: we need to pass the rootStruct (which should contain s as delegate) here
+ if err := bind(element, rootStruct); err != nil {
+ return nil, errors.Wrap(err, "error binding")
+ }
+ }
+ }
+ return s, nil
+}
+
+func (s *ServiceAccessPoint) SapRequest(apdu readWriteModel.APDU) error {
+ log.Debug().Msgf("SapRequest(%d)\n%s", s.serviceID, apdu)
+
+ if s.serviceElement == nil {
+ return errors.New("unbound service access point")
+ }
+ return s.serviceElement.Indication(apdu)
+}
+
+func (s *ServiceAccessPoint) SapIndication(readWriteModel.APDU, []byte) error {
+ panic("this should be implemented by outer struct")
+}
+
+func (s *ServiceAccessPoint) SapResponse(apdu readWriteModel.APDU) error {
+ log.Debug().Msgf("SapResponse(%d)\n%s", s.serviceID, apdu)
+
+ if s.serviceElement == nil {
+ return errors.New("unbound service access point")
+ }
+ return s.serviceElement.Confirmation(apdu)
+}
+
+func (s *ServiceAccessPoint) SapConfirmation(readWriteModel.APDU, []byte) error {
+ panic("this should be implemented by outer struct")
+}
+
+func (s *ServiceAccessPoint) _setServiceElement(serviceElement _ApplicationServiceElement) {
+ s.serviceElement = serviceElement
+}
+
+// _ApplicationServiceElement is a interface used for documentation
+type _ApplicationServiceElement interface {
+ Request(apdu readWriteModel.APDU) error
+ Indication(apdu readWriteModel.APDU) error
+ Response(apdu readWriteModel.APDU) error
+ Confirmation(apdu readWriteModel.APDU) error
+ _setElementService(elementService _ServiceAccessPoint)
+}
+
+type ApplicationServiceElement struct {
+ elementID *int
+ elementService _ServiceAccessPoint
+}
+
+func NewApplicationServiceElement(aseID *int, rootStruct _ApplicationServiceElement) (*ApplicationServiceElement, error) {
+ a := &ApplicationServiceElement{
+ elementID: aseID,
+ }
+
+ if aseID != nil {
+ if _, ok := elementMap[*aseID]; ok {
+ return nil, errors.Errorf("already an application service element %d", *aseID)
+ }
+ elementMap[*aseID] = a
+
+ // automatically bind
+ if service, ok := serviceMap[*aseID]; ok {
+ if service.serviceElement != nil {
+ return nil, errors.Errorf("service access point %d already bound", *aseID)
+ }
+
+ // Note: we need to pass the rootStruct (which should contain a as delegate) here
+ if err := bind(rootStruct, service); err != nil {
+ return nil, errors.Wrap(err, "error binding")
+ }
+ }
+ }
+ return a, nil
+}
+
+func (a *ApplicationServiceElement) Request(apdu readWriteModel.APDU) error {
+ log.Debug().Msgf("Request\n%s", apdu)
+
+ if a.elementService == nil {
+ return errors.New("unbound application service element")
+ }
+
+ return a.elementService.SapIndication(apdu, nil) // TODO: where to get the source from
+}
+
+func (a *ApplicationServiceElement) Indication(apdu readWriteModel.APDU) error {
+ panic("this should be implemented by outer struct")
+}
+
+func (a *ApplicationServiceElement) Response(apdu readWriteModel.APDU) error {
+ log.Debug().Msgf("Response\n%s", apdu)
+
+ if a.elementService == nil {
+ return errors.New("unbound application service element")
+ }
+
+ return a.elementService.SapConfirmation(apdu, nil) // TODO: where to get the source from
+}
+
+func (a *ApplicationServiceElement) Confirmation(apdu readWriteModel.APDU) error {
+ panic("this should be implemented by outer struct")
+}
+
+func (a *ApplicationServiceElement) _setElementService(elementService _ServiceAccessPoint) {
+ a.elementService = elementService
+}
+
+// bind a list of clients and servers together, top down
+func bind(args ...interface{}) error {
+ // generic bind is pairs of names
+ if len(args) == 0 {
+ // find unbound clients and bind them
+ for cid, client := range clientMap {
+ // skip those that are already bound
+ if client.clientPeer != nil {
+ continue
+ }
+
+ server, ok := serverMap[cid]
+ if !ok {
+ return errors.Errorf("unmatched server %d", cid)
+ }
+
+ if server.serverPeer != nil {
+ return errors.Errorf("server already bound %d", cid)
+ }
+
+ if err := bind(client, server); err != nil {
+ return errors.Wrap(err, "error binding")
+ }
+ }
+
+ // see if there are any unbound servers
+ for sid, server := range serverMap {
+ if server.serverPeer != nil {
+ continue
+ }
+
+ if _, ok := clientMap[sid]; !ok {
+ return errors.Errorf("unmatched client %d", sid)
+ } else {
+ return errors.Errorf("unknown unbound server %d", sid)
+ }
+ }
+
+ // find unbound application service elements and bind them
+ for eid, element := range elementMap {
+ // skip those that are already bound
+ if element.elementService != nil {
+ continue
+ }
+
+ service, ok := serviceMap[eid]
+ if !ok {
+ return errors.Errorf("unmatched element %d", eid)
+ }
+
+ if service.serviceElement == nil {
+ return errors.Errorf("element already bound %d", eid)
+ }
+
+ if err := bind(element, service); err != nil {
+ return errors.Wrap(err, "error binding")
+ }
+ }
+
+ // see if there are any unbound services
+ for sid, service := range serviceMap {
+ if service.serviceElement != nil {
+ continue
+ }
+
+ if _, ok := elementMap[sid]; !ok {
+ return errors.Errorf("unmatched service %d", sid)
+ } else {
+ return errors.Errorf("unknown unbound service %d", sid)
+ }
+ }
+ }
+
+ // go through the argument pairs
+ for i := 0; i < len(args); i++ {
+ client := args[i]
+ log.Debug().Msgf("client %v", client)
+ server := args[i+1]
+ log.Debug().Msgf("server %v", server)
+
+ // make sure we're binding clients and servers
+ clientCast, okClient := client.(_Client)
+ serverCast, okServer := server.(_Server)
+ elementServiceCast, okElementService := client.(_ApplicationServiceElement)
+ serviceAccessPointCast, okServiceAccessPoint := server.(_ServiceAccessPoint)
+ if okClient && okServer {
+ clientCast._setClientPeer(serverCast)
+ serverCast._setServerPeer(clientCast)
+ } else if okElementService && okServiceAccessPoint { // we could be binding application clients and servers
+ elementServiceCast._setElementService(serviceAccessPointCast)
+ serviceAccessPointCast._setServiceElement(elementServiceCast)
+ } else {
+ return errors.New("bind() requires a client and a server")
+ }
+ }
+ log.Debug().Msg("bound")
+ return nil
+}
diff --git a/plc4go/internal/bacnetip/Task.go b/plc4go/internal/bacnetip/Task.go
new file mode 100644
index 0000000000..4771f6bdd4
--- /dev/null
+++ b/plc4go/internal/bacnetip/Task.go
@@ -0,0 +1,44 @@
+/*
+ * 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 "time"
+
+// TODO: this is a placeholder for a tasking framework
+type _Task struct {
+ taskTime time.Time
+ isScheduled bool
+}
+
+func (t *_Task) installTask(when *time.Time, delta *time.Duration) {
+ // TODO: schedule task
+}
+
+func (t *_Task) suspendTask() {
+ // TODO: suspend task
+}
+
+func (t *_Task) resume() {
+ // TODO: resume task
+}
+
+type OneShotTask struct {
+ _Task
+}
diff --git a/plc4go/internal/bacnetip/TransactionStateMachine.go b/plc4go/internal/bacnetip/TransactionStateMachine.go
index c4132e3744..2298b8f231 100644
--- a/plc4go/internal/bacnetip/TransactionStateMachine.go
+++ b/plc4go/internal/bacnetip/TransactionStateMachine.go
@@ -20,11 +20,9 @@
package bacnetip
import (
- "bytes"
"context"
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"
"time"
@@ -141,1724 +139,3 @@ func (t *TransactionStateMachine) handleOutboundMessage(message spi.Message) (ha
return false, nil
}
}
-
-// TODO: this is a placeholder for a tasking framework
-type _Task struct {
- taskTime time.Time
- isScheduled bool
-}
-
-func (t *_Task) installTask(when *time.Time, delta *time.Duration) {
- // TODO: schedule task
-}
-
-func (t *_Task) suspendTask() {
- // TODO: suspend task
-}
-
-func (t *_Task) resume() {
- // TODO: resume task
-}
-
-type OneShotTask struct {
- _Task
-}
-
-// TODO: this is the interface to the outside for the SSM // TODO: maybe we should port that as non interface first
-type ServiceAccessPoint interface {
- GetDeviceInventory() *DeviceInventory
- GetLocalDevice() DeviceEntry
- GetProposedWindowSize() uint8
- Request(apdu readWriteModel.APDU)
- SapRequest(apdu readWriteModel.APDU)
- SapResponse(apdu readWriteModel.APDU)
- // TODO: wrap that properly
- GetClientTransactions() []interface{}
- // TODO: wrap that properly
- GetServerTransactions() []interface{}
- GetApplicationTimeout() *uint
-}
-
-// TODO: interface to client // TODO: maybe we should port that as non interface first
-type Client interface {
- request(apdu readWriteModel.APDU)
- confirmation(apdu readWriteModel.APDU)
-}
-
-type SSMState uint8
-
-const (
- IDLE SSMState = iota
- SEGMENTED_REQUEST
- AWAIT_CONFIRMATION
- AWAIT_RESPONSE
- SEGMENTED_RESPONSE
- SEGMENTED_CONFIRMATION
- COMPLETED
- ABORTED
-)
-
-func (s SSMState) String() string {
- switch s {
- case IDLE:
- return "IDLE"
- case SEGMENTED_REQUEST:
- return "SEGMENTED_REQUEST"
- case AWAIT_CONFIRMATION:
- return "AWAIT_CONFIRMATION"
- case AWAIT_RESPONSE:
- return "AWAIT_RESPONSE"
- case SEGMENTED_RESPONSE:
- return "SEGMENTED_RESPONSE"
- case SEGMENTED_CONFIRMATION:
- return "SEGMENTED_CONFIRMATION"
- case COMPLETED:
- return "COMPLETED"
- case ABORTED:
- return "ABORTED"
- default:
- return "Unknown"
- }
-}
-
-type segmentAPDU struct {
- originalApdu readWriteModel.APDU
- originalInvokeId uint8
- serviceBytes []byte
- serviceChoice readWriteModel.BACnetConfirmedServiceChoice
- isAck bool
-}
-
-// SSM - Segmentation State Machine
-type SSM struct {
- OneShotTask
-
- ssmSAP ServiceAccessPoint
-
- pduAddress []byte
- deviceEntry *DeviceEntry
-
- invokeId uint8
-
- state SSMState
- segmentAPDU *segmentAPDU // TODO: rename that to segmentAPDUSource or something
- segmentSize uint
- segmentCount uint8
-
- retryCount uint
- segmentRetryCount uint
- sentAllSegments bool
- lastSequenceNumber uint8
- initialSequenceNumber uint8
- actualWindowSize *uint8
-
- numberOfApduRetries uint
- apduTimeout uint
- segmentationSupported readWriteModel.BACnetSegmentation
- segmentTimeout uint
- maxSegmentsAccepted *readWriteModel.MaxSegmentsAccepted
- maxApduLengthAccepted *readWriteModel.MaxApduLengthAccepted
-}
-
-func NewSSM(sap ServiceAccessPoint, pduAddress []byte) (SSM, error) {
- log.Debug().Interface("sap", sap).Bytes("pdu_address", pduAddress).Msg("init")
- deviceEntry, err := sap.GetDeviceInventory().getEntryForDestination(pduAddress)
- if err != nil {
- return SSM{}, errors.Wrap(err, "Can't create SSM")
- }
- localDevice := sap.GetLocalDevice()
- return SSM{
- ssmSAP: sap,
- pduAddress: pduAddress,
- deviceEntry: deviceEntry,
- state: IDLE,
- numberOfApduRetries: localDevice.NumberOfAPDURetries,
- apduTimeout: localDevice.APDUTimeout,
- segmentationSupported: localDevice.SegmentationSupported,
- segmentTimeout: localDevice.APDUSegmentTimeout,
- maxSegmentsAccepted: localDevice.MaxSegmentsAccepted,
- maxApduLengthAccepted: localDevice.MaximumApduLengthAccepted,
- }, nil
-}
-
-func (s *SSM) startTimer(millis uint) {
- log.Debug().Msgf("Start timer %d", millis)
- s.restartTimer(millis)
-}
-
-func (s *SSM) stopTimer() {
- log.Debug().Msg("Stop Timer")
- if s.isScheduled {
- log.Debug().Msg("is scheduled")
- s.suspendTask()
- }
-}
-
-func (s *SSM) restartTimer(millis uint) {
- log.Debug().Msgf("restartTimer %d", millis)
- if s.isScheduled {
- log.Debug().Msg("is scheduled")
- s.suspendTask()
- }
-
- delta := time.Millisecond * time.Duration(millis)
- s.installTask(nil, &delta)
-}
-
-// setState This function is called when the derived class wants to change state
-func (s *SSM) setState(newState SSMState, timer *uint) error {
- log.Debug().Msgf("setState %s timer=%d", newState, timer)
- if s.state == COMPLETED || s.state == ABORTED {
- return errors.Errorf("Invalid state transition from %s to %s", s.state, newState)
- }
-
- s.stopTimer()
-
- s.state = newState
-
- if timer != nil {
- s.startTimer(*timer)
- }
- return nil
-}
-
-// setSegmentationContext This function is called to set the segmentation context
-func (s *SSM) setSegmentationContext(apdu readWriteModel.APDU) error {
- log.Debug().Msgf("setSegmentationContext\n%s", apdu)
- switch apdu := apdu.(type) {
- case readWriteModel.APDUConfirmedRequestExactly:
- if apdu.GetSegmentedMessage() || apdu.GetMoreFollows() {
- return errors.New("Can't handle already segmented message")
- }
- bytes, err := apdu.GetServiceRequest().Serialize()
- if err != nil {
- return errors.Wrap(err, "Can serialize service request")
- }
- segmentAPDU := segmentAPDU{
- originalApdu: apdu,
- originalInvokeId: apdu.GetInvokeId(),
- serviceBytes: bytes,
- serviceChoice: apdu.GetServiceRequest().GetServiceChoice(),
- }
- s.segmentAPDU = &segmentAPDU
- case readWriteModel.APDUComplexAckExactly:
- if apdu.GetSegmentedMessage() || apdu.GetMoreFollows() {
- return errors.New("Can't handle already segmented message")
- }
- bytes, err := apdu.GetServiceAck().Serialize()
- if err != nil {
- return errors.Wrap(err, "Can serialize service request")
- }
- segmentAPDU := segmentAPDU{
- originalApdu: apdu,
- serviceBytes: bytes,
- serviceChoice: apdu.GetServiceAck().GetServiceChoice(),
- isAck: true,
- }
- s.segmentAPDU = &segmentAPDU
- default:
- return errors.Errorf("invalid APDU type %T", apdu)
- }
- return nil
-}
-
-// 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) {
- log.Debug().Msgf("Get segment %d", index)
- if s.segmentAPDU == nil {
- return nil, false, errors.New("No segment apdu set")
- }
-
- if index > s.segmentCount {
- return nil, false, errors.Errorf("Invalid segment number %d, APDU has %d segments", index, s.segmentCount)
- }
-
- // 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
- }
-
- moreFollows = index < s.segmentCount-1
- sequenceNumber := index % 255
- proposedWindowSize := s.actualWindowSize
- if index == 0 {
- getProposedWindowSize := s.ssmSAP.GetProposedWindowSize()
- proposedWindowSize = &getProposedWindowSize
- }
- log.Debug().Msgf("proposedWindowSize %d", proposedWindowSize)
- serviceChoice := &s.segmentAPDU.serviceChoice
- offset := uint(index) * s.segmentSize
- segmentBytes := s.segmentAPDU.serviceBytes[offset : offset+s.segmentSize]
- if !s.segmentAPDU.isAck {
- 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(
- true,
- moreFollows,
- segmentedResponseAccepted,
- *s.maxSegmentsAccepted,
- *s.maxApduLengthAccepted,
- s.segmentAPDU.originalInvokeId,
- &sequenceNumber,
- proposedWindowSize,
- nil,
- serviceChoice,
- segmentBytes,
- 0,
- )
- } else {
- log.Debug().Msg("complex ack context")
- segmentAPDU = readWriteModel.NewAPDUComplexAck(
- true,
- moreFollows,
- s.segmentAPDU.originalInvokeId,
- &sequenceNumber,
- proposedWindowSize,
- nil,
- serviceChoice,
- segmentBytes,
- 0,
- )
- }
- return segmentAPDU, moreFollows, nil
-}
-
-// 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 readWriteModel.APDU) error {
- log.Debug().Msgf("appendSegment\n%s", apdu)
- switch apdu := apdu.(type) {
- case readWriteModel.APDUConfirmedRequestExactly:
- if apdu.GetSegmentedMessage() || apdu.GetMoreFollows() {
- return errors.New("Can't handle already segmented message")
- }
- bytes, err := apdu.GetServiceRequest().Serialize()
- if err != nil {
- return errors.Wrap(err, "Can serialize service request")
- }
- s.segmentAPDU.serviceBytes = append(s.segmentAPDU.serviceBytes, bytes...)
- case readWriteModel.APDUComplexAckExactly:
- if apdu.GetSegmentedMessage() || apdu.GetMoreFollows() {
- return errors.New("Can't handle already segmented message")
- }
- bytes, err := apdu.GetServiceAck().Serialize()
- if err != nil {
- return errors.Wrap(err, "Can serialize service request")
- }
- s.segmentAPDU.serviceBytes = append(s.segmentAPDU.serviceBytes, bytes...)
- default:
- return errors.Errorf("invalid APDU type %T", apdu)
- }
- return nil
-}
-
-func (s *SSM) inWindow(sequenceA, sequenceB uint8) bool {
- log.Debug().Msgf("inWindow %d-%d", sequenceA, sequenceB)
- return (uint(sequenceA)-uint(sequenceB)-256)%256 < uint(*s.actualWindowSize)
-}
-
-func (s *SSM) fillWindow(sequenceNumber uint8) error {
- log.Debug().Msgf("fillWindow %d", sequenceNumber)
- for i := uint8(0); i < *s.actualWindowSize; i++ {
- apdu, moreFollows, err := s.getSegment(sequenceNumber + i)
- if err != nil {
- return errors.Wrapf(err, "Error sending out segment %d", i)
- }
- s.ssmSAP.Request(apdu)
- if moreFollows {
- s.sentAllSegments = true
- }
- }
- return nil
-}
-
-type ClientSSM struct {
- SSM
-}
-
-func NewClientSSM(sap ServiceAccessPoint, pduAddress []byte) (*ClientSSM, error) {
- log.Debug().Interface("sap", sap).Bytes("pduAddress", pduAddress).Msg("init")
- ssm, err := NewSSM(sap, pduAddress)
- if err != nil {
- return nil, err
- }
- // TODO: if deviceEntry is not there get it now...
- if ssm.deviceEntry == nil {
- // TODO: get entry for device, store it in inventory
- log.Debug().Msg("Accquire device information")
- }
- return &ClientSSM{
- SSM: ssm,
- }, nil
-}
-
-// 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)
- // do the regular state change
- if err := s.SSM.setState(newState, timer); err != nil {
- return errors.Wrap(err, "error during SSM state transition")
- }
-
- if s.state == COMPLETED || s.state == ABORTED {
- log.Debug().Msg("remove from active transaction")
- s.ssmSAP.GetClientTransactions() // TODO remove "this" transaction from the list
- if s.deviceEntry == nil {
- // TODO: release device entry
- log.Debug().Msg("release device entry")
- }
- }
- return nil
-}
-
-// request This function is called by client transaction functions when it wants to send a message to the device
-func (s *ClientSSM) request(apdu readWriteModel.APDU) {
- 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
- s.ssmSAP.Request(apdu)
-}
-
-// TODO: maybe use another name for that
-// 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 readWriteModel.APDU) error {
- log.Debug().Msgf("indication\n%s", apdu)
- // make sure we're getting confirmed requests
- var apduConfirmedRequest readWriteModel.APDUConfirmedRequest
- if apdu, ok := apdu.(readWriteModel.APDUConfirmedRequestExactly); !ok {
- return errors.Errorf("Invalid APDU type %T", apdu)
- }
-
- // save the request and set the segmentation context
- if err := s.setSegmentationContext(apdu); 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.deviceEntry == nil || s.deviceEntry.MaximumApduLengthAccepted != nil {
- s.segmentSize = uint(s.maxApduLengthAccepted.NumberOfOctets())
- } else if s.deviceEntry.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())
- } else {
- s.segmentSize = utils.Min(*s.deviceEntry.MaximumNpduLength, uint(s.maxApduLengthAccepted.NumberOfOctets()))
- }
- log.Debug().Msgf("segment size %d", s.segmentSize)
-
- s.invokeId = apduConfirmedRequest.GetInvokeId()
- log.Debug().Msgf("invoke ID: %d", s.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)
- if more > 0 {
- s.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 {
- log.Debug().Msg("local device can't send segmented requests")
- abort, err := s.abort(readWriteModel.BACnetAbortReason_SEGMENTATION_NOT_SUPPORTED)
- if err != nil {
- return errors.Wrap(err, "Error creating abort")
- }
- s.response(abort)
- return nil
- }
-
- if s.deviceEntry == nil {
- log.Debug().Msg("no server info for segmentation support")
- } else if s.deviceEntry.SegmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_TRANSMIT && s.deviceEntry.SegmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_BOTH {
- log.Debug().Msg("server can't receive segmented requests")
- abort, err := s.abort(readWriteModel.BACnetAbortReason_SEGMENTATION_NOT_SUPPORTED)
- if err != nil {
- return errors.Wrap(err, "Error creating abort")
- }
- s.response(abort)
- return nil
- }
-
- // make sure we don't exceed the number of segments in our request that the server said it was willing to accept
- if s.deviceEntry == nil {
- log.Debug().Msg("no server info for maximum number of segments")
- } else if s.deviceEntry.MaxSegmentsAccepted == nil {
- log.Debug().Msgf("server doesn't say maximum number of segments")
- } else if s.segmentCount > s.deviceEntry.MaxSegmentsAccepted.MaxSegments() {
- log.Debug().Msg("server can't receive enough segments")
- abort, err := s.abort(readWriteModel.BACnetAbortReason_APDU_TOO_LONG)
- if err != nil {
- return errors.Wrap(err, "Error creating abort")
- }
- s.response(abort)
- return nil
- }
- }
-
- // send out the first segment (or the whole thing)
- if s.segmentCount == 1 {
- // unsegmented
- s.sentAllSegments = true
- s.retryCount = 0
- if err := s.setState(AWAIT_CONFIRMATION, &s.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(SEGMENTED_REQUEST, &s.segmentTimeout); err != nil {
- return errors.Wrap(err, "error switching state")
- }
- }
-
- // deliver to the device
- segment, _, err := s.getSegment(0)
- if err != nil {
- return errors.Wrap(err, "error getting segment")
- }
- s.request(segment)
- return nil
-}
-
-// response This function is called by client transaction functions when they want to send a message to the application.
-func (s *ClientSSM) response(apdu readWriteModel.APDU) {
- log.Debug().Msgf("response\n%s", apdu)
- // make sure it has a good source and destination
- // TODO: check if source == s.pduAddress
- // TODO: check if
-
- // send it to the application
- s.ssmSAP.SapResponse(apdu)
-}
-
-// confirmation This function is called by the device for all upstream messages related to the transaction.
-func (s *ClientSSM) confirmation(apdu readWriteModel.APDU) error {
- log.Debug().Msgf("confirmation\n%s", apdu)
-
- switch s.state {
- case SEGMENTED_REQUEST:
- return s.segmentedRequest(apdu)
- case AWAIT_CONFIRMATION:
- return s.awaitConfirmation(apdu)
- case SEGMENTED_CONFIRMATION:
- return s.segmentedConfirmation(apdu)
- default:
- return errors.Errorf("Invalid state %s", s.state)
- }
-}
-
-// processTask This function is called when something has taken too long
-func (s *ClientSSM) processTask() error {
- log.Debug().Msg("processTask")
- switch s.state {
- case SEGMENTED_REQUEST:
- return s.segmentedRequestTimeout()
- case AWAIT_CONFIRMATION:
- return s.awaitConfirmationTimeout()
- case SEGMENTED_CONFIRMATION:
- return s.segmentedConfirmationTimeout()
- case COMPLETED, ABORTED:
- return nil
- default:
- return errors.Errorf("Invalid state %s", s.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)
-
- // change the state to aborted
- if err := s.setState(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)
- // return it
- return abortApdu, nil
-}
-
-// segmentedRequest This function is called when the client is sending a segmented request and receives an apdu
-func (s *ClientSSM) segmentedRequest(apdu readWriteModel.APDU) error {
- log.Debug().Msgf("segmentedRequest\n%s", apdu)
-
- 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
-
- // duplicate ack received?
- if !s.inWindow(apdu.GetSequenceNumber(), s.initialSequenceNumber) {
- log.Debug().Msg("not in window")
- s.restartTimer(s.segmentTimeout)
- } else if s.sentAllSegments {
- log.Debug().Msg("all done sending request")
-
- if err := s.setState(AWAIT_CONFIRMATION, &s.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 {
- return errors.Wrap(err, "error filling window")
- }
- s.restartTimer(s.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 err != nil {
- return errors.Wrap(err, "error creating abort")
- }
- s.request(abort) // send it ot the device
- s.response(abort) // send it ot the application
- } else {
- if err := s.setState(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 err != nil {
- return errors.Wrap(err, "error creating abort")
- }
- s.request(abort) // send it ot the device
- s.response(abort) // send it ot the application
- } else if !apdu.GetSegmentedMessage() {
- // ack is not segmented
- if err := s.setState(COMPLETED, nil); err != nil {
- return errors.Wrap(err, "error switching state")
- }
- s.response(apdu)
- } else {
- // set the segmented response context
- if err := s.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(SEGMENTED_CONFIRMATION, &s.segmentTimeout); err != nil {
- return errors.Wrap(err, "error switching state")
- }
- }
- case readWriteModel.APDUErrorExactly:
- log.Debug().Msg("error/reject/abort")
- if err := s.setState(COMPLETED, nil); err != nil {
- return errors.Wrap(err, "error switching state")
- }
- s.response(apdu)
- default:
- return errors.Errorf("Invalid apdu %T", apdu)
- }
- return nil
-}
-
-func (s *ClientSSM) segmentedRequestTimeout() error {
- log.Debug().Msg("segmentedRequestTimeout")
-
- // Try again
- if s.segmentRetryCount < s.numberOfApduRetries {
- log.Debug().Msg("retry segmented request")
- s.segmentRetryCount++
- s.startTimer(s.segmentTimeout)
-
- if s.initialSequenceNumber == 0 {
- apdu, _, err := s.getSegment(0)
- if err != nil {
- return errors.Wrap(err, "error getting first segment")
- }
- s.request(apdu)
- } else {
- if err := s.fillWindow(s.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
- if err != nil {
- return errors.Wrap(err, "error creating abort")
- }
- s.response(abort)
- }
- return nil
-}
-
-func (s *ClientSSM) awaitConfirmation(apdu readWriteModel.APDU) error {
- log.Debug().Msgf("awaitConfirmation\n%s", apdu)
-
- switch apdu := apdu.(type) {
- case readWriteModel.APDUAbortExactly:
- log.Debug().Msg("Server aborted")
-
- if err := s.setState(ABORTED, nil); err != nil {
- return errors.Wrap(err, "error switching state")
- }
- s.response(apdu)
- case readWriteModel.APDUSimpleAckExactly, readWriteModel.APDUErrorExactly, readWriteModel.APDURejectExactly:
- log.Debug().Msg("simple ack, error or reject")
-
- if err := s.setState(COMPLETED, nil); err != nil {
- return errors.Wrap(err, "error switching state")
- }
- s.response(apdu)
- case readWriteModel.APDUComplexAckExactly:
- log.Debug().Msg("complex ack")
-
- if !apdu.GetSegmentedMessage() {
- log.Debug().Msg("unsegmented")
-
- if err := s.setState(COMPLETED, nil); err != nil {
- return errors.Wrap(err, "error switching state")
- }
- s.response(apdu)
- } else if s.segmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_RECEIVE && s.segmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_BOTH {
- log.Debug().Msg("local device can't receive segmented messages")
-
- abort, err := s.abort(readWriteModel.BACnetAbortReason_SEGMENTATION_NOT_SUPPORTED)
- if err != nil {
- return errors.Wrap(err, "error creating abort")
- }
- s.response(abort)
- } else if *apdu.GetSequenceNumber() == 0 {
- log.Debug().Msg("segmented response")
-
- // set the segmented response context
- if err := s.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(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)
- s.request(segmentAck)
- } else {
- log.Debug().Msg("Invalid apdu in this state")
-
- abort, err := s.abort(readWriteModel.BACnetAbortReason_INVALID_APDU_IN_THIS_STATE)
- if err != nil {
- return errors.Wrap(err, "error creating abort")
- }
- s.request(abort) // send it to the device
- s.response(abort) // send it to the application
- }
- case readWriteModel.APDUSegmentAckExactly:
- log.Debug().Msg("segment ack(!?)")
- s.restartTimer(s.segmentTimeout)
- default:
- return errors.Errorf("invalid apdu %T", apdu)
- }
- return nil
-}
-
-func (s *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++
-
- // 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
- return err
- }
- s.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
- if err != nil {
- return errors.Wrap(err, "error creating abort")
- }
- s.response(abort)
- }
- return nil
-}
-
-func (s *ClientSSM) segmentedConfirmation(apdu readWriteModel.APDU) error {
- log.Debug().Msgf("segmentedConfirmation\n%s", 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)
- if err != nil {
- return errors.Wrap(err, "error creating abort")
- }
- s.request(abort) // send it to the device
- s.response(abort) // send it to the application
- }
-
- // it must be segmented
- if !apduComplexAck.GetSegmentedMessage() {
- abort, err := s.abort(readWriteModel.BACnetAbortReason_INVALID_APDU_IN_THIS_STATE)
- if err != nil {
- return errors.Wrap(err, "error creating abort")
- }
- s.request(abort) // send it to the device
- s.response(abort) // send it to the application
- }
-
- // 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)
-
- // segment received out of order
- s.restartTimer(s.segmentTimeout)
- segmentAck := readWriteModel.NewAPDUSegmentAck(true, false, s.invokeId, s.initialSequenceNumber, *s.actualWindowSize, 0)
- s.request(segmentAck)
- return nil
- }
-
- // add the data
- if err := s.appendSegment(apdu); err != nil {
- return errors.Wrap(err, "error appending the segment")
- }
-
- // update the sequence number
- s.lastSequenceNumber = s.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)
- s.request(segmentAck)
-
- if err := s.setState(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)))
- if err != nil {
- return errors.Wrap(err, "error parsing apdu")
- }
- s.response(parse)
- } else if *apduComplexAck.GetSequenceNumber() == s.initialSequenceNumber+*s.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)
- s.request(segmentAck)
- } else {
- log.Debug().Msg("Wait for more segments")
-
- s.restartTimer(s.segmentTimeout)
- }
-
- return nil
-}
-
-func (s *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
- if err != nil {
- return errors.Wrap(err, "error creating abort")
- }
- s.response(abort)
- return nil
-}
-
-type ServerSSM struct {
- SSM
- segmentedResponseAccepted bool
-}
-
-func NewServerSSM(sap ServiceAccessPoint, pduAddress []byte) (*ServerSSM, error) {
- log.Debug().Interface("sap", sap).Bytes("pduAddress", pduAddress).Msg("init")
- ssm, err := NewSSM(sap, pduAddress)
- if err != nil {
- return nil, err
- }
- // TODO: if deviceEntry is not there get it now...
- if &ssm.deviceEntry == nil {
- // TODO: get entry for device, store it in inventory
- log.Debug().Msg("Accquire device information")
- }
- return &ServerSSM{
- SSM: ssm,
- segmentedResponseAccepted: true,
- }, nil
-}
-
-// setState This function is called when the client wants to change state
-func (s *ServerSSM) setState(newState SSMState, timer *uint) error {
- log.Debug().Msgf("setState %s timer=%d", newState, timer)
- // do the regular state change
- if err := s.SSM.setState(newState, timer); err != nil {
- return errors.Wrap(err, "error during SSM state transition")
- }
-
- if s.state == COMPLETED || s.state == ABORTED {
- log.Debug().Msg("remove from active transaction")
- s.ssmSAP.GetServerTransactions() // TODO remove "this" transaction from the list
- if s.deviceEntry != nil {
- // TODO: release device entry
- log.Debug().Msg("release device entry")
- }
- }
- return nil
-}
-
-// request This function is called by transaction functions to send to the application
-func (s *ServerSSM) request(apdu readWriteModel.APDU) {
- 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
- s.ssmSAP.SapRequest(apdu)
-}
-
-// TODO: maybe use another name for that
-// indication This function is called for each downstream packet related to
-// the transaction
-func (s *ServerSSM) indication(apdu readWriteModel.APDU) error {
- log.Debug().Msgf("indication\n%s", apdu)
- // make sure we're getting confirmed requests
-
- switch s.state {
- case IDLE:
- return s.idle(apdu)
- case SEGMENTED_REQUEST:
- return s.segmentedRequest(apdu)
- case AWAIT_RESPONSE:
- return s.awaitResponse(apdu)
- case SEGMENTED_RESPONSE:
- return s.segmentedResponse(apdu)
- default:
- return errors.Errorf("invalid state %s", s.state)
- }
-}
-
-// response This function is called by client transaction functions when they want to send a message to the application.
-func (s *ServerSSM) response(apdu readWriteModel.APDU) {
- log.Debug().Msgf("response\n%s", apdu)
- // make sure it has a good source and destination
- // TODO: check if source == none
- // TODO: check if destnation = s.pduAddress
-
- // send it via the device
- s.ssmSAP.Request(apdu)
-}
-
-// 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 readWriteModel.APDU) error {
- log.Debug().Msgf("confirmation\n%s", apdu)
-
- // check to see we are in the correct state
- if s.state != AWAIT_RESPONSE {
- log.Debug().Msg("warning: no expecting a response")
- }
-
- switch apdu := apdu.(type) {
- // abort response
- case readWriteModel.APDUAbortExactly:
- log.Debug().Msg("abort")
-
- if err := s.setState(ABORTED, nil); err != nil {
- return errors.Wrap(err, "Error setting state to aborted")
- }
-
- // end the response to the device
- s.response(apdu)
- return nil
- // simple response
- case readWriteModel.APDUSimpleAckExactly, readWriteModel.APDUErrorExactly, readWriteModel.APDURejectExactly:
- log.Debug().Msg("simple ack, error or reject")
-
- // transaction completed
- if err := s.setState(COMPLETED, nil); err != nil {
- return errors.Wrap(err, "Error setting state to aborted")
- }
-
- // send the response to the device
- s.response(apdu)
- return nil
- // complex ack
- case readWriteModel.APDUComplexAckExactly:
- log.Debug().Msg("complex ack")
-
- // save the response and set the segmentation context
- if err := s.setSegmentationContext(apdu); err != nil {
- return errors.Wrap(err, "error settings segmentation context")
- }
-
- // the segment size is the minimum of the size of the largest packet that can be delivered to the client and the
- // largest it can accept
- if s.deviceEntry == nil || s.deviceEntry.MaximumNpduLength == nil {
- s.segmentSize = uint(s.maxApduLengthAccepted.NumberOfOctets())
- } else {
- s.segmentSize = utils.Min(*s.deviceEntry.MaximumNpduLength, uint(s.maxApduLengthAccepted.NumberOfOctets()))
- }
-
- // compute the segment count
- if len(apdu.GetSegment()) == 0 {
- // always at least one segment
- s.segmentCount = 1
- } else {
- // split into chunks, maybe need one more
- var segmentCount, more int
- segmentCount, more = len(s.segmentAPDU.serviceBytes)/int(s.segmentSize), len(s.segmentAPDU.serviceBytes)%int(s.segmentSize)
- if more > 0 {
- s.segmentCount += 1
- }
- log.Debug().Msgf("segment count: %d", segmentCount)
-
- // make sure we support segmented transmit if we need to
- if s.segmentCount > 1 {
- log.Debug().Msgf("segmentation required, %d segments", s.segmentCount)
-
- // make sure we support segmented transmit
- if s.segmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_TRANSMIT && s.segmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_BOTH {
- log.Debug().Msg("server can't send segmented requests")
- abort, err := s.abort(readWriteModel.BACnetAbortReason_SEGMENTATION_NOT_SUPPORTED)
- if err != nil {
- return errors.Wrap(err, "Error creating abort")
- }
- s.response(abort)
- return nil
- }
-
- // make sure client supports segmented receive
- if !s.segmentedResponseAccepted {
- log.Debug().Msg("client can't receive segmented responses")
- abort, err := s.abort(readWriteModel.BACnetAbortReason_SEGMENTATION_NOT_SUPPORTED)
- if err != nil {
- return errors.Wrap(err, "Error creating abort")
- }
- s.response(abort)
- return nil
- }
-
- // make sure we don't exceed the number of segments in our response that the client said it was willing to accept
- // in the request
- if s.maxSegmentsAccepted != nil && s.segmentCount > s.maxSegmentsAccepted.MaxSegments() {
- log.Debug().Msg("client can't receive enough segments")
- 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
- if err != nil {
- return errors.Wrap(err, "Error creating abort")
- }
- s.response(abort)
- return nil
- }
- }
-
- // initialize the state
- s.segmentRetryCount = 0
- s.initialSequenceNumber = 0
- s.actualWindowSize = nil
-
- // send out the first segment (or the whole thing)
- if s.segmentCount == 1 {
- s.response(apdu)
- if err := s.setState(COMPLETED, nil); err != nil {
- return errors.Wrap(err, "Error setting state to aborted")
- }
- } else {
- segment, _, err := s.getSegment(0)
- if err != nil {
- return errors.Wrap(err, "error getting first segment")
- }
- s.response(segment)
- if err := s.setState(SEGMENTED_RESPONSE, nil); err != nil {
- return errors.Wrap(err, "Error setting state to aborted")
- }
- }
- }
- default:
- return errors.Errorf("Invalid APDU %T", apdu)
- }
- return nil
-}
-
-// processTask This function is called when the client has failed to send all the segments of a segmented request,
-// the application has taken too long to complete the request, or the client failed to ack the segments of a
-// segmented response
-func (s *ServerSSM) processTask() error {
- log.Debug().Msg("processTask")
- switch s.state {
- case SEGMENTED_REQUEST:
- return s.segmentedRequestTimeout()
- case AWAIT_CONFIRMATION:
- return s.awaitResponseTimeout()
- case SEGMENTED_CONFIRMATION:
- return s.segmentedResponseTimeout()
- case COMPLETED, ABORTED:
- return nil
- default:
- return errors.Errorf("Invalid state %s", s.state)
- }
-}
-
-// abort This function is called when the transaction should be aborted
-func (s *ServerSSM) abort(reason readWriteModel.BACnetAbortReason) (readWriteModel.APDU, error) {
- log.Debug().Msgf("abort\n%s", reason)
-
- // change the state to aborted
- if err := s.setState(ABORTED, nil); err != nil {
- return nil, errors.Wrap(err, "Error setting state to aborted")
- }
-
- // build an abort PDU to return
- abortApdu := readWriteModel.NewAPDUAbort(true, s.invokeId, readWriteModel.NewBACnetAbortReasonTagged(reason, uint32(reason), 0), 0)
- // return it
- return abortApdu, nil
-}
-
-func (s *ServerSSM) idle(apdu readWriteModel.APDU) error {
- log.Debug().Msgf("idle %s", apdu)
-
- // make sure we're getting confirmed requests
- var apduConfirmedRequest readWriteModel.APDUConfirmedRequest
- if apdu, ok := apdu.(readWriteModel.APDUConfirmedRequestExactly); !ok {
- return errors.Errorf("Invalid APDU type %T", apdu)
- }
-
- // save the invoke ID
- s.invokeId = apduConfirmedRequest.GetInvokeId()
- log.Debug().Msgf("invoke ID: %d", s.invokeId)
-
- // remember if the client accepts segmented responses
- s.segmentedResponseAccepted = apduConfirmedRequest.GetSegmentedResponseAccepted()
-
- // if there is a cache record, check to see if it needs to be updated
- if apduConfirmedRequest.GetSegmentedResponseAccepted() && s.deviceEntry != nil {
- switch s.deviceEntry.SegmentationSupported {
- case readWriteModel.BACnetSegmentation_NO_SEGMENTATION:
- log.Debug().Msg("client actually supports segmented receive")
- s.deviceEntry.SegmentationSupported = readWriteModel.BACnetSegmentation_SEGMENTED_RECEIVE
-
- // TODO: bacpypes updates the cache here but as we have a pointer to the entry we should need that. Maybe we should because concurrency... lets see later
- case readWriteModel.BACnetSegmentation_SEGMENTED_TRANSMIT:
- log.Debug().Msg("client actually supports both segmented transmit and receive")
- s.deviceEntry.SegmentationSupported = readWriteModel.BACnetSegmentation_SEGMENTED_BOTH
-
- // TODO: bacpypes updates the cache here but as we have a pointer to the entry we should need that. Maybe we should because concurrency... lets see later
- case readWriteModel.BACnetSegmentation_SEGMENTED_RECEIVE, readWriteModel.BACnetSegmentation_SEGMENTED_BOTH:
- // all good
- default:
- return errors.New("invalid segmentation supported in device info")
- }
- }
-
- // decode the maximum that the client can receive in one APDU, and if there is a value in the device information then
- // use that one because it came from reading device object property value or from an I-Am message that was
- // received
- getMaxApduLengthAccepted := apduConfirmedRequest.GetMaxApduLengthAccepted()
- s.maxApduLengthAccepted = &getMaxApduLengthAccepted
- if s.deviceEntry != nil && s.deviceEntry.MaximumApduLengthAccepted != nil {
- if *s.deviceEntry.MaximumApduLengthAccepted < *s.maxApduLengthAccepted {
- log.Debug().Msg("apdu max reponse encoding error")
- } else {
- s.maxApduLengthAccepted = s.deviceEntry.MaximumApduLengthAccepted
- }
- }
- log.Debug().Msgf("maxApduLengthAccepted %s", *s.maxApduLengthAccepted)
-
- // save the number of segments the client is willing to accept in the ack, if this is None then the value is unknown or more than 64
- getMaxSegmentsAccepted := apduConfirmedRequest.GetMaxSegmentsAccepted()
- s.maxSegmentsAccepted = &getMaxSegmentsAccepted
-
- // unsegmented request
- if len(apduConfirmedRequest.GetSegment()) <= 0 {
- if err := s.setState(AWAIT_RESPONSE, nil); err != nil {
- return errors.Wrap(err, "Error setting state to aborted")
- }
- s.request(apdu)
- return nil
- }
-
- // make sure we support segmented requests
- if s.segmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_RECEIVE && s.segmentationSupported != readWriteModel.BACnetSegmentation_SEGMENTED_BOTH {
- abort, err := s.abort(readWriteModel.BACnetAbortReason_SEGMENTATION_NOT_SUPPORTED)
- if err != nil {
- return errors.Wrap(err, "error creating abort")
- }
- s.response(abort)
- return nil
- }
-
- // save the response and set the segmentation context
- if err := s.setSegmentationContext(apdu); err != nil {
- return errors.Wrap(err, "error settings segmentation context")
- }
-
- // the window size is the minimum of what I would propose and what the device has proposed
- minWindowSize := utils.Min(*apduConfirmedRequest.GetProposedWindowSize(), s.ssmSAP.GetProposedWindowSize())
- s.actualWindowSize = &minWindowSize
- log.Debug().Msgf("actualWindowSize? min(%d, %d) -> %d", apduConfirmedRequest.GetProposedWindowSize(), s.ssmSAP.GetProposedWindowSize(), s.actualWindowSize)
-
- // initialize the state
- s.lastSequenceNumber = 0
- s.initialSequenceNumber = 0
- if err := s.setState(SEGMENTED_REQUEST, &s.segmentTimeout); err != nil {
- return errors.Wrap(err, "Error setting state to aborted")
- }
-
- // send back a segment ack
- segack := readWriteModel.NewAPDUSegmentAck(false, true, s.invokeId, s.initialSequenceNumber, *s.actualWindowSize, 0)
- log.Debug().Msgf("segAck: %s", segack)
- s.response(segack)
- return nil
-}
-
-func (s *ServerSSM) segmentedRequest(apdu readWriteModel.APDU) error {
- log.Debug().Msgf("segmentedRequest\n%s", apdu)
-
- // some kind of problem
- if _, ok := apdu.(readWriteModel.APDUAbortExactly); ok {
- if err := s.setState(COMPLETED, nil); err != nil {
- return errors.Wrap(err, "Error setting state to aborted")
- }
- s.response(apdu)
- return nil
- }
-
- // the only messages we should be getting are confirmed requests
- var apduConfirmedRequest readWriteModel.APDUConfirmedRequest
- if castedApdu, ok := apdu.(readWriteModel.APDUConfirmedRequestExactly); !ok {
- abort, err := s.abort(readWriteModel.BACnetAbortReason_INVALID_APDU_IN_THIS_STATE)
- if err != nil {
- return errors.Wrap(err, "error creating abort")
- }
- s.request(abort) // send it ot the device
- s.response(abort) // send it ot the application
- } else {
- apduConfirmedRequest = castedApdu
- }
-
- // it must be segmented
- if !apduConfirmedRequest.GetSegmentedMessage() {
- abort, err := s.abort(readWriteModel.BACnetAbortReason_INVALID_APDU_IN_THIS_STATE)
- if err != nil {
- return errors.Wrap(err, "error creating abort")
- }
- s.request(abort) // send it ot the device
- s.response(abort) // send it ot the application
- }
-
- // proper segment number
- if *apduConfirmedRequest.GetSequenceNumber() != s.lastSequenceNumber+1 {
- log.Debug().Msgf("segment %d received out of order, should be %d", *apduConfirmedRequest.GetSequenceNumber(), s.lastSequenceNumber+1)
-
- // segment received out of order
- s.restartTimer(s.segmentTimeout)
-
- // send back a segment ack
- segack := readWriteModel.NewAPDUSegmentAck(true, true, s.invokeId, s.initialSequenceNumber, *s.actualWindowSize, 0)
- s.response(segack)
- return nil
- }
-
- // add the data
- if err := s.appendSegment(apdu); err != nil {
- return errors.Wrap(err, "error appending segment")
- }
-
- // update the sequence number
- s.lastSequenceNumber++
-
- // last segment?
- if !apduConfirmedRequest.GetMoreFollows() {
- log.Debug().Msg("No more follows")
-
- // send back the final segment ack
- segack := readWriteModel.NewAPDUSegmentAck(false, true, s.invokeId, s.lastSequenceNumber, *s.actualWindowSize, 0)
- s.response(segack)
-
- // forward the whole thing to the application
- if err := s.setState(AWAIT_RESPONSE, s.ssmSAP.GetApplicationTimeout()); err != nil {
- return errors.Wrap(err, "Error setting state to aborted")
- }
- // TODO: here we need to rebuild again yada yada
- // 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)))
- if err != nil {
- return errors.Wrap(err, "error parsing apdu")
- }
- s.request(parse)
- } else if *apduConfirmedRequest.GetSequenceNumber() == s.initialSequenceNumber+*s.actualWindowSize {
- log.Debug().Msg("last segment in the group")
-
- s.initialSequenceNumber = s.lastSequenceNumber
- s.restartTimer(s.segmentTimeout)
-
- // send back a segment ack
- segack := readWriteModel.NewAPDUSegmentAck(false, true, s.invokeId, s.initialSequenceNumber, *s.actualWindowSize, 0)
- s.response(segack)
- } else {
- // wait for more segments
- s.restartTimer(s.segmentTimeout)
- }
-
- return nil
-}
-
-func (s *ServerSSM) segmentedRequestTimeout() error {
- log.Debug().Msg("segmentedRequestTimeout")
-
- // give up
- if err := s.setState(ABORTED, nil); err != nil {
- return errors.Wrap(err, "Error setting state to aborted")
- }
- return nil
-}
-
-func (s *ServerSSM) awaitResponse(apdu readWriteModel.APDU) error {
- log.Debug().Msgf("awaitResponse\n%s", apdu)
-
- switch apdu.(type) {
- case readWriteModel.APDUConfirmedRequestExactly:
- log.Debug().Msg("client is trying this request again")
- case readWriteModel.APDUAbortExactly:
- log.Debug().Msg("client aborting this request")
-
- // forward to the application
- if err := s.setState(ABORTED, nil); err != nil {
- return errors.Wrap(err, "Error setting state to aborted")
- }
- s.request(apdu)
- default:
- return errors.Errorf("invalid APDU %T", apdu)
- }
- return nil
-}
-
-// awaitResponseTimeout This function is called when the application has taken too long to respond to a clients request.
-// The client has probably long since given up
-func (s *ServerSSM) awaitResponseTimeout() error {
- log.Debug().Msg("awaitResponseTimeout")
-
- abort, err := s.abort(readWriteModel.BACnetAbortReason(64)) // Note: this is a proprietary code used by bacpypes for server timeout. We just use that here too to keep consistent
- if err != nil {
- return errors.Wrap(err, "error creating abort")
- }
- s.request(abort)
- return nil
-}
-
-func (s *ServerSSM) segmentedResponse(apdu readWriteModel.APDU) error {
- log.Debug().Msgf("segmentedResponse\n%s", apdu)
-
- // client is ready for the next segment
- switch apdu := apdu.(type) {
- case readWriteModel.APDUSegmentAckExactly:
- log.Debug().Msg("segment ack")
-
- // actual window size is provided by client
- getActualWindowSize := apdu.GetActualWindowSize()
- s.actualWindowSize = &getActualWindowSize
-
- // duplicate ack received?
- if !s.inWindow(apdu.GetSequenceNumber(), s.initialSequenceNumber) {
- log.Debug().Msg("not in window")
- s.restartTimer(s.segmentTimeout)
- } else if s.sentAllSegments {
- // final ack received?
- log.Debug().Msg("all done sending response")
- if err := s.setState(COMPLETED, nil); err != nil {
- return errors.Wrap(err, "Error setting state to aborted")
- }
- } else {
- log.Debug().Msg("more segments to send")
-
- s.initialSequenceNumber = apdu.GetSequenceNumber() + 1
- actualWindowSize := apdu.GetActualWindowSize()
- s.actualWindowSize = &actualWindowSize
- s.segmentRetryCount = 0
- if err := s.fillWindow(s.initialSequenceNumber); err != nil {
- return errors.Wrap(err, "error filling window")
- }
- s.restartTimer(s.segmentRetryCount)
- }
- // some kind of problem
- case readWriteModel.APDUAbortExactly:
- if err := s.setState(COMPLETED, nil); err != nil {
- return errors.Wrap(err, "Error setting state to aborted")
- }
- s.response(apdu)
- default:
- return errors.Errorf("Invalid APDU %T", apdu)
- }
- return nil
-}
-
-func (s *ServerSSM) segmentedResponseTimeout() error {
- log.Debug().Msg("segmentedResponseTimeout")
-
- // try again
- if s.segmentRetryCount < s.numberOfApduRetries {
- s.segmentRetryCount++
- s.startTimer(s.segmentTimeout)
- if err := s.fillWindow(s.initialSequenceNumber); err != nil {
- return errors.Wrap(err, "error filling window")
- }
- } else {
- // five up
- if err := s.setState(ABORTED, nil); err != nil {
- return errors.Wrap(err, "Error setting state to aborted")
- }
- }
- return nil
-}
-
-type StateMachineAccessPoint struct {
- Client
- ServiceAccessPoint
-
- localDevice DeviceEntry
- deviceInventory *DeviceInventory
- nextInvokeId uint8
- clientTransactions []*ClientSSM
- serverTransactions []*ServerSSM
- numberOfApduRetries int
- apduTimeout int
- maxApduLengthAccepted int
- segmentationSupported readWriteModel.BACnetSegmentation
- segmentTimeout int
- maxSegmentsAccepted int
- proposedWindowSize int
- dccEnableDisable readWriteModel.BACnetConfirmedServiceRequestDeviceCommunicationControlEnableDisable
- applicationTimeout int
-}
-
-func NewStateMachineAccessPoint(localDevice DeviceEntry, deviceInventory *DeviceInventory, sapID int, cid int) StateMachineAccessPoint {
- log.Debug().Msgf("NewStateMachineAccessPoint localDevice=%v deviceInventory=%v sap=%v cid=%v", localDevice, deviceInventory, sapID, cid)
-
- // basic initialization
- // TODO: init client
- // TODO: init sap
- return StateMachineAccessPoint{
- // save a reference to the device information cache
- localDevice: localDevice,
- deviceInventory: deviceInventory,
-
- // client settings
- nextInvokeId: 1,
- clientTransactions: nil,
-
- // server settings
- serverTransactions: nil,
-
- // confirmed request defaults
- numberOfApduRetries: 3,
- apduTimeout: 3000,
- maxApduLengthAccepted: 1024,
-
- // segmentation defaults
- segmentationSupported: readWriteModel.BACnetSegmentation_NO_SEGMENTATION,
- segmentTimeout: 1500,
- maxSegmentsAccepted: 2,
- proposedWindowSize: 2,
-
- // device communication control
- dccEnableDisable: readWriteModel.BACnetConfirmedServiceRequestDeviceCommunicationControlEnableDisable_ENABLE,
-
- // how long the state machine is willing to wait for the application
- // layer to form a response and send it
- applicationTimeout: 3000,
- }
-}
-
-// getNextInvokeId Called by clients to get an unused invoke ID
-func (s *StateMachineAccessPoint) getNextInvokeId(address []byte) (uint8, error) {
- log.Debug().Msg("getNextInvokeId")
-
- initialID := s.nextInvokeId
- for {
- invokeId := s.nextInvokeId
- s.nextInvokeId++
-
- // see if we've checked for them all
- if initialID == s.nextInvokeId {
- return 0, errors.New("No available invoke ID")
- }
-
- if len(s.clientTransactions) == 0 {
- return invokeId, nil
- }
-
- // TODO: double check that the logic here is right
- for _, tr := range s.clientTransactions {
- if invokeId == tr.invokeId && bytes.Equal(address, tr.pduAddress) {
- return invokeId, nil
- }
- }
- }
-}
-
-// confirmation Packets coming up the stack are APDU's
-func (s *StateMachineAccessPoint) confirmation(apdu readWriteModel.APDU, pduSource []byte) error {
- log.Debug().Msgf("confirmation\n%s", apdu)
-
- // check device communication control
- switch s.dccEnableDisable {
- case readWriteModel.BACnetConfirmedServiceRequestDeviceCommunicationControlEnableDisable_ENABLE:
- log.Debug().Msg("communications enabled")
- case readWriteModel.BACnetConfirmedServiceRequestDeviceCommunicationControlEnableDisable_DISABLE:
- switch {
- case apdu.GetApduType() == 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 &&
- apdu.(readWriteModel.APDUConfirmedRequest).GetServiceRequest().GetServiceChoice() == readWriteModel.BACnetConfirmedServiceChoice_REINITIALIZE_DEVICE:
- log.Debug().Msg("continue with reinitialize device")
- case apdu.GetApduType() == readWriteModel.ApduType_UNCONFIRMED_REQUEST_PDU &&
- apdu.(readWriteModel.APDUUnconfirmedRequest).GetServiceRequest().GetServiceChoice() == readWriteModel.BACnetUnconfirmedServiceChoice_WHO_IS:
- log.Debug().Msg("continue with Who-Is")
- default:
- log.Debug().Msg("not a Who-Is, dropped")
- return nil
- }
- case readWriteModel.BACnetConfirmedServiceRequestDeviceCommunicationControlEnableDisable_DISABLE_INITIATION:
- log.Debug().Msg("initiation disabled")
- }
-
- 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) {
- tr = serverTransactionElement
- break
- }
- }
- if tr == nil {
- // build a server transaction
- var err error
- tr, err = NewServerSSM(s, pduSource)
- if err != nil {
- return errors.Wrap(err, "Error building server ssm")
- }
- s.serverTransactions = append(s.serverTransactions, tr)
- }
-
- // let it run with the apdu
- if err := tr.indication(apdu); err != nil {
- return errors.Wrap(err, "error runnning indication")
- }
- case readWriteModel.APDUUnconfirmedRequestExactly:
- // deliver directly to the application
- s.SapRequest(apdu)
- case readWriteModel.APDUSimpleAckExactly, readWriteModel.APDUComplexAckExactly, readWriteModel.APDUErrorExactly, readWriteModel.APDURejectExactly:
- // 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) {
- break
- }
- }
- if tr == nil {
- // TODO: log at least
- return nil
- }
-
- // send the packet on to the transaction
- if err := tr.confirmation(apdu); err != nil {
- return errors.Wrap(err, "error running confirmation")
- }
- case readWriteModel.APDUAbortExactly:
- // find the transaction being aborted
- if apdu.GetServer() {
- var tr *ClientSSM
- for _, tr := range s.clientTransactions {
- if apdu.(interface{ GetOriginalInvokeId() uint8 }).GetOriginalInvokeId() == tr.invokeId && bytes.Equal(pduSource, tr.pduAddress) {
- break
- }
- }
- if tr == nil {
- // TODO: log at least
- return nil
- }
-
- // send the packet on to the transaction
- if err := tr.confirmation(apdu); err != nil {
- return errors.Wrap(err, "error running confirmation")
- }
- } else {
- var tr *ServerSSM
- for _, serverTransactionElement := range s.serverTransactions {
- if apdu.GetOriginalInvokeId() == serverTransactionElement.invokeId && bytes.Equal(pduSource, serverTransactionElement.pduAddress) {
- tr = serverTransactionElement
- break
- }
- }
- if tr == nil {
- // TODO: log at least
- return nil
- }
-
- // send the packet on to the transaction
- if err := tr.indication(apdu); err != nil {
- return errors.Wrap(err, "error running indication")
- }
- }
- case readWriteModel.APDUSegmentAckExactly:
- // find the transaction being aborted
- if apdu.GetServer() {
- var tr *ClientSSM
- for _, tr := range s.clientTransactions {
- if apdu.(interface{ GetOriginalInvokeId() uint8 }).GetOriginalInvokeId() == tr.invokeId && bytes.Equal(pduSource, tr.pduAddress) {
- break
- }
- }
- if tr == nil {
- // TODO: log at least
- return nil
- }
-
- // send the packet on to the transaction
- if err := tr.confirmation(apdu); err != nil {
- return errors.Wrap(err, "error running confirmation")
- }
- } else {
- var tr *ServerSSM
- for _, serverTransactionElement := range s.serverTransactions {
- if apdu.GetOriginalInvokeId() == serverTransactionElement.invokeId && bytes.Equal(pduSource, serverTransactionElement.pduAddress) {
- tr = serverTransactionElement
- break
- }
- }
- if tr == nil {
- // TODO: log at least
- return nil
- }
-
- // send the packet on to the transaction
- if err := tr.indication(apdu); err != nil {
- return errors.Wrap(err, "error running indication")
- }
- }
- default:
- return errors.Errorf("invalid APDU %T", apdu)
- }
- return nil
-}
-
-// sapIndication This function is called when the application is requesting a new transaction as a client.
-func (s *StateMachineAccessPoint) sapIndication(apdu readWriteModel.APDU, pduDestination []byte) error {
- log.Debug().Msgf("sapIndication\n%s", apdu)
-
- // check device communication control
- switch s.dccEnableDisable {
- case readWriteModel.BACnetConfirmedServiceRequestDeviceCommunicationControlEnableDisable_ENABLE:
- log.Debug().Msg("communications enabled")
- case readWriteModel.BACnetConfirmedServiceRequestDeviceCommunicationControlEnableDisable_DISABLE:
- log.Debug().Msg("communications disabled")
- return nil
- case readWriteModel.BACnetConfirmedServiceRequestDeviceCommunicationControlEnableDisable_DISABLE_INITIATION:
- log.Debug().Msg("initiation disabled")
- if apdu.GetApduType() == readWriteModel.ApduType_UNCONFIRMED_REQUEST_PDU && apdu.(readWriteModel.APDUUnconfirmedRequest).GetServiceRequest().GetServiceChoice() == readWriteModel.BACnetUnconfirmedServiceChoice_I_AM {
- log.Debug().Msg("continue with I-Am")
- } else {
- log.Debug().Msg("not an I-Am")
- return nil
- }
- }
-
- switch apdu := apdu.(type) {
- case readWriteModel.APDUUnconfirmedRequestExactly:
- // deliver to the device
- s.request(apdu)
- case readWriteModel.APDUConfirmedRequestExactly:
- // make sure it has an invoke ID
- // 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) {
- return errors.New("invoke ID in use")
- }
- }
-
- // warning for bogus requests
- // TODO: not sure if we have that or if it is relvant (localstationaddr)
-
- // create a client transaction state machine
- tr, err := NewClientSSM(s, pduDestination)
- if err != nil {
- return errors.Wrap(err, "error creating client ssm")
- }
-
- // add it to our transactions to track it
- s.clientTransactions = append(s.clientTransactions, tr)
-
- // let it run
- if err := tr.indication(apdu); err != nil {
- return errors.Wrap(err, "error doing indication")
- }
- default:
- return errors.Errorf("invalid APDU %T", apdu)
- }
-
- return nil
-}
-
-// 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 readWriteModel.APDU, pduDestination []byte) error {
- log.Debug().Msgf("sapConfirmation\n%s", apdu)
- 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) {
- break
- }
- }
- if tr == nil {
- // TODO: log at least
- return nil
- }
-
- // pass control to the transaction
- if err := tr.confirmation(apdu); err != nil {
- return errors.Wrap(err, "error running confirmation")
- }
- default:
- return errors.Errorf("invalid APDU %T", apdu)
- }
- return nil
-}
diff --git a/plc4go/internal/bacnetip/TransactionStateMachine_test.go b/plc4go/internal/bacnetip/TransactionStateMachine_test.go
deleted file mode 100644
index 715bdc495f..0000000000
--- a/plc4go/internal/bacnetip/TransactionStateMachine_test.go
+++ /dev/null
@@ -1,432 +0,0 @@
-/*
- * 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 (
- "context"
- readWriteModel "github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model"
- "github.com/apache/plc4x/plc4go/spi"
- "github.com/apache/plc4x/plc4go/spi/transports/test"
- "reflect"
- "testing"
- "time"
-)
-
-func TestTransactionStateMachine_Expect(t1 *testing.T) {
- type fields struct {
- MessageCodec *MessageCodec
- deviceInventory *DeviceInventory
- retryCount int
- segmentRetryCount int
- duplicateCount int
- sentAllSegments bool
- lastSequenceNumber int
- initialSequenceNumber int
- actualWindowSize int
- proposeWindowSize int
- segmentTimer int
- RequestTimer int
- }
- type args struct {
- ctx context.Context
- acceptsMessage spi.AcceptsMessage
- handleMessage spi.HandleMessage
- handleError spi.HandleError
- ttl time.Duration
- }
- tests := []struct {
- name string
- fields fields
- args args
- wantErr bool
- }{
- // TODO: Add test cases.
- }
- for _, tt := range tests {
- t1.Run(tt.name, func(t1 *testing.T) {
- t := &TransactionStateMachine{
- MessageCodec: tt.fields.MessageCodec,
- deviceInventory: tt.fields.deviceInventory,
- retryCount: tt.fields.retryCount,
- segmentRetryCount: tt.fields.segmentRetryCount,
- duplicateCount: tt.fields.duplicateCount,
- sentAllSegments: tt.fields.sentAllSegments,
- lastSequenceNumber: tt.fields.lastSequenceNumber,
- initialSequenceNumber: tt.fields.initialSequenceNumber,
- actualWindowSize: tt.fields.actualWindowSize,
- proposeWindowSize: tt.fields.proposeWindowSize,
- segmentTimer: tt.fields.segmentTimer,
- RequestTimer: tt.fields.RequestTimer,
- }
- if err := t.Expect(tt.args.ctx, tt.args.acceptsMessage, tt.args.handleMessage, tt.args.handleError, tt.args.ttl); (err != nil) != tt.wantErr {
- t1.Errorf("Expect() error = %v, wantErr %v", err, tt.wantErr)
- }
- })
- }
-}
-
-func TestTransactionStateMachine_GetCodec(t1 *testing.T) {
- type fields struct {
- MessageCodec *MessageCodec
- deviceInventory *DeviceInventory
- retryCount int
- segmentRetryCount int
- duplicateCount int
- sentAllSegments bool
- lastSequenceNumber int
- initialSequenceNumber int
- actualWindowSize int
- proposeWindowSize int
- segmentTimer int
- RequestTimer int
- }
- tests := []struct {
- name string
- fields fields
- want spi.MessageCodec
- }{
- // TODO: Add test cases.
- }
- for _, tt := range tests {
- t1.Run(tt.name, func(t1 *testing.T) {
- t := &TransactionStateMachine{
- MessageCodec: tt.fields.MessageCodec,
- deviceInventory: tt.fields.deviceInventory,
- retryCount: tt.fields.retryCount,
- segmentRetryCount: tt.fields.segmentRetryCount,
- duplicateCount: tt.fields.duplicateCount,
- sentAllSegments: tt.fields.sentAllSegments,
- lastSequenceNumber: tt.fields.lastSequenceNumber,
- initialSequenceNumber: tt.fields.initialSequenceNumber,
- actualWindowSize: tt.fields.actualWindowSize,
- proposeWindowSize: tt.fields.proposeWindowSize,
- segmentTimer: tt.fields.segmentTimer,
- RequestTimer: tt.fields.RequestTimer,
- }
- if got := t.GetCodec(); !reflect.DeepEqual(got, tt.want) {
- t1.Errorf("GetCodec() = %v, want %v", got, tt.want)
- }
- })
- }
-}
-
-func TestTransactionStateMachine_Send(t1 *testing.T) {
- type fields struct {
- MessageCodec *MessageCodec
- deviceInventory *DeviceInventory
- retryCount int
- segmentRetryCount int
- duplicateCount int
- sentAllSegments bool
- lastSequenceNumber int
- initialSequenceNumber int
- actualWindowSize int
- proposeWindowSize int
- segmentTimer int
- RequestTimer int
- }
- type args struct {
- message spi.Message
- }
- tests := []struct {
- name string
- fields fields
- args args
- wantErr bool
- }{
- // TODO: Add test cases.
- }
- for _, tt := range tests {
- t1.Run(tt.name, func(t1 *testing.T) {
- t := &TransactionStateMachine{
- MessageCodec: tt.fields.MessageCodec,
- deviceInventory: tt.fields.deviceInventory,
- retryCount: tt.fields.retryCount,
- segmentRetryCount: tt.fields.segmentRetryCount,
- duplicateCount: tt.fields.duplicateCount,
- sentAllSegments: tt.fields.sentAllSegments,
- lastSequenceNumber: tt.fields.lastSequenceNumber,
- initialSequenceNumber: tt.fields.initialSequenceNumber,
- actualWindowSize: tt.fields.actualWindowSize,
- proposeWindowSize: tt.fields.proposeWindowSize,
- segmentTimer: tt.fields.segmentTimer,
- RequestTimer: tt.fields.RequestTimer,
- }
- if err := t.Send(tt.args.message); (err != nil) != tt.wantErr {
- t1.Errorf("Send() error = %v, wantErr %v", err, tt.wantErr)
- }
- })
- }
-}
-
-func TestTransactionStateMachine_SendRequest(t1 *testing.T) {
- type fields struct {
- MessageCodec *MessageCodec
- deviceInventory *DeviceInventory
- retryCount int
- segmentRetryCount int
- duplicateCount int
- sentAllSegments bool
- lastSequenceNumber int
- initialSequenceNumber int
- actualWindowSize int
- proposeWindowSize int
- segmentTimer int
- RequestTimer int
- }
- type args struct {
- ctx context.Context
- message spi.Message
- acceptsMessage spi.AcceptsMessage
- handleMessage spi.HandleMessage
- handleError spi.HandleError
- ttl time.Duration
- }
- tests := []struct {
- name string
- fields fields
- args args
- wantErr bool
- }{
- // TODO: Add test cases.
- }
- for _, tt := range tests {
- t1.Run(tt.name, func(t1 *testing.T) {
- t := &TransactionStateMachine{
- MessageCodec: tt.fields.MessageCodec,
- deviceInventory: tt.fields.deviceInventory,
- retryCount: tt.fields.retryCount,
- segmentRetryCount: tt.fields.segmentRetryCount,
- duplicateCount: tt.fields.duplicateCount,
- sentAllSegments: tt.fields.sentAllSegments,
- lastSequenceNumber: tt.fields.lastSequenceNumber,
- initialSequenceNumber: tt.fields.initialSequenceNumber,
- actualWindowSize: tt.fields.actualWindowSize,
- proposeWindowSize: tt.fields.proposeWindowSize,
- segmentTimer: tt.fields.segmentTimer,
- RequestTimer: tt.fields.RequestTimer,
- }
- if err := t.SendRequest(tt.args.ctx, tt.args.message, tt.args.acceptsMessage, tt.args.handleMessage, tt.args.handleError, tt.args.ttl); (err != nil) != tt.wantErr {
- t1.Errorf("SendRequest() error = %v, wantErr %v", err, tt.wantErr)
- }
- })
- }
-}
-
-func TestTransactionStateMachine_handleOutboundMessage(t1 *testing.T) {
- type fields struct {
- MessageCodec *MessageCodec
- deviceInventory *DeviceInventory
- retryCount int
- segmentRetryCount int
- duplicateCount int
- sentAllSegments bool
- lastSequenceNumber int
- initialSequenceNumber int
- actualWindowSize int
- proposeWindowSize int
- segmentTimer int
- RequestTimer int
- }
- type args struct {
- message spi.Message
- }
- tests := []struct {
- name string
- fields fields
- args args
- wantHandled bool
- wantErr bool
- }{
- {
- name: "message not relevant",
- },
- {
- name: "Normal unsgemented message",
- fields: fields{
- MessageCodec: NewMessageCodec(test.NewTransportInstance(test.NewTransport())),
- deviceInventory: nil,
- retryCount: 0,
- segmentRetryCount: 0,
- duplicateCount: 0,
- sentAllSegments: false,
- lastSequenceNumber: 0,
- initialSequenceNumber: 0,
- actualWindowSize: 0,
- proposeWindowSize: 0,
- segmentTimer: 0,
- RequestTimer: 0,
- },
- args: args{
- message: readWriteModel.NewBVLCOriginalUnicastNPDU(
- readWriteModel.NewNPDU(
- 1,
- readWriteModel.NewNPDUControl(false, false, false, false, readWriteModel.NPDUNetworkPriority_NORMAL_MESSAGE),
- nil,
- nil,
- nil,
- nil,
- nil,
- nil,
- nil,
- nil,
- readWriteModel.NewAPDUComplexAck(
- false,
- false,
- 13,
- nil,
- nil,
- readWriteModel.NewBACnetServiceAckReadProperty(
- readWriteModel.CreateBACnetContextTagObjectIdentifier(0, 2, 1),
- readWriteModel.CreateBACnetPropertyIdentifierTagged(1, 85),
- nil,
- readWriteModel.NewBACnetConstructedDataAnalogValuePresentValue(
- readWriteModel.CreateBACnetApplicationTagReal(101),
- readWriteModel.CreateBACnetOpeningTag(3),
- readWriteModel.CreateBACnetTagHeaderBalanced(true, 3, 3),
- readWriteModel.CreateBACnetClosingTag(3),
- 3,
- nil,
- ),
- 0,
- ),
- nil,
- nil,
- 0,
- ),
- 0,
- ),
- 0,
- ),
- },
- },
- {
- name: "Normal segmented message",
- fields: fields{
- MessageCodec: NewMessageCodec(test.NewTransportInstance(test.NewTransport())),
- deviceInventory: func() *DeviceInventory {
- var deviceInventory = DeviceInventory{
- devices: map[string]DeviceEntry{
- "123": {
- DeviceIdentifier: nil,
- MaximumApduLengthAccepted: func() *readWriteModel.MaxApduLengthAccepted {
- x := readWriteModel.MaxApduLengthAccepted_NUM_OCTETS_206
- return &x
- }(),
- SegmentationSupported: readWriteModel.BACnetSegmentation_SEGMENTED_BOTH,
- VendorId: 0,
- DeviceObjects: nil,
- },
- },
- }
- return &deviceInventory
- }(),
- retryCount: 0,
- segmentRetryCount: 0,
- duplicateCount: 0,
- sentAllSegments: false,
- lastSequenceNumber: 0,
- initialSequenceNumber: 0,
- actualWindowSize: 0,
- proposeWindowSize: 0,
- segmentTimer: 0,
- RequestTimer: 0,
- },
- args: args{
- message: readWriteModel.NewBVLCOriginalUnicastNPDU(
- readWriteModel.NewNPDU(
- 1,
- readWriteModel.NewNPDUControl(false, true, false, false, readWriteModel.NPDUNetworkPriority_NORMAL_MESSAGE),
- nil,
- func() *uint8 {
- var elements uint8 = 3
- return &elements
- }(),
- []uint8{0x31, 0x32, 0x33},
- nil,
- nil,
- nil,
- nil,
- nil,
- readWriteModel.NewAPDUComplexAck(
- false,
- false,
- 13,
- nil,
- nil,
- readWriteModel.NewBACnetServiceAckReadProperty(
- readWriteModel.CreateBACnetContextTagObjectIdentifier(0, 2, 1),
- readWriteModel.CreateBACnetPropertyIdentifierTagged(1, 85),
- nil,
- readWriteModel.NewBACnetConstructedDataActionText(
- readWriteModel.CreateBACnetApplicationTagUnsignedInteger(100),
- func() []readWriteModel.BACnetApplicationTagCharacterString {
- var characterStrings []readWriteModel.BACnetApplicationTagCharacterString
- for i := 0; i < 100; i++ {
- characterStrings = append(characterStrings, readWriteModel.CreateBACnetApplicationTagCharacterString(readWriteModel.BACnetCharacterEncoding_ISO_10646, "ALAAARM!!"))
- }
- return characterStrings
- }(),
- readWriteModel.CreateBACnetOpeningTag(3),
- readWriteModel.CreateBACnetTagHeaderBalanced(true, 3, 3),
- readWriteModel.CreateBACnetClosingTag(3),
- 3,
- nil,
- ),
- 0,
- ),
- nil,
- nil,
- 0,
- ),
- 0,
- ),
- 0,
- ),
- },
- },
- }
- for _, tt := range tests {
- t1.Run(tt.name, func(t1 *testing.T) {
- t := &TransactionStateMachine{
- MessageCodec: tt.fields.MessageCodec,
- deviceInventory: tt.fields.deviceInventory,
- retryCount: tt.fields.retryCount,
- segmentRetryCount: tt.fields.segmentRetryCount,
- duplicateCount: tt.fields.duplicateCount,
- sentAllSegments: tt.fields.sentAllSegments,
- lastSequenceNumber: tt.fields.lastSequenceNumber,
- initialSequenceNumber: tt.fields.initialSequenceNumber,
- actualWindowSize: tt.fields.actualWindowSize,
- proposeWindowSize: tt.fields.proposeWindowSize,
- segmentTimer: tt.fields.segmentTimer,
- RequestTimer: tt.fields.RequestTimer,
- }
- gotHandled, err := t.handleOutboundMessage(tt.args.message)
- if (err != nil) != tt.wantErr {
- t1.Errorf("handleOutboundMessage() error = %v, wantErr %v", err, tt.wantErr)
- return
- }
- if gotHandled != tt.wantHandled {
- t1.Errorf("handleOutboundMessage() gotHandled = %v, want %v", gotHandled, tt.wantHandled)
- }
- })
- }
-}