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 2026/08/18 20:22:23 UTC
(plc4x) 05/17: fix(plc4go/eip): port the plc4j connect handshake and un-skip the driver testsuite (GH-954)
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 3fbfd792cb018675e1ceec02506331f670468e58
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Tue Aug 18 16:55:20 2026 +0200
fix(plc4go/eip): port the plc4j connect handshake and un-skip the driver testsuite (GH-954)
Rewrites Connect/setupConnection into listServices -> registerSession ->
probeAttributes -> openConnectionManager, mirroring plc4j's
doConnectHandshake/probeAttributes/openConnectionManager. The background
setup goroutine (used when awaitSetupComplete is false, as in the driver
testsuite) now runs on context.WithoutCancel(ctx) instead of the caller's
ctx, which used to get canceled the moment GetConnection returned - that
was the "really strange bug" behind the long-standing test skip.
Adds Requests.go with sendRequestAndWait/sendTransactedRequestAndWait,
shared helpers for the handshake and (later) the read/write paths. Close
now also sends a ForwardClose ahead of UnregisterSession. As agreed in the
spec, a failed ForwardOpen falls back to unconnected mode rather than
failing the connect, unlike plc4j.
Unskipping the testsuite exposed three latent bugs it happened to never
exercise before:
- MessageCodec.Receive never drained the test transport's staged bytes
(GetNumBytesAvailableInBuffer doesn't pull from the channel by itself);
fixed with the same FillBuffer-first pattern already used by S7/Modbus.
- AnsiExtendedSymbolSegment's generated Serialize wrote a zero-filled tag
name because its WriteString call carried no encoding, so the shared
byte-buffer writer matched no case and only zero-padded; added
WithEncoding("UTF8").
- plcTag's XML serialization was missing the PlcTagItem/tag wrapper the
golden testsuite expects (matching plc4j's PlcTagItem), so read-response
round-trips never matched; added the wrapper, scoped to the eip package.
---
plc4go/internal/eip/Connection.go | 443 ++++++++++-----------
plc4go/internal/eip/MessageCodec.go | 16 +
plc4go/internal/eip/Requests.go | 102 +++++
plc4go/internal/eip/Tag.go | 14 +
.../readwrite/model/AnsiExtendedSymbolSegment.go | 3 +-
plc4go/tests/drivers/tests/eip_driver_test.go | 1 -
6 files changed, 340 insertions(+), 239 deletions(-)
diff --git a/plc4go/internal/eip/Connection.go b/plc4go/internal/eip/Connection.go
index 341befe276..13451a5eaf 100644
--- a/plc4go/internal/eip/Connection.go
+++ b/plc4go/internal/eip/Connection.go
@@ -131,19 +131,26 @@ func (c *Connection) Connect(ctx context.Context) error {
// For testing purposes we can skip the waiting for a complete connection
if !c.driverContext.awaitSetupComplete {
+ // The caller's ctx is canceled as soon as GetConnection returns - the
+ // background handshake must not inherit that cancellation (GH-954).
+ setupCtx := context.WithoutCancel(ctx)
c.wg.Go(func() {
- if err := c.setupConnection(ctx); err != nil {
+ setupCtx, cancel := utils.WithNamedTimeout(setupCtx, "eip setup timeout", 10*time.Second)
+ defer cancel()
+ if err := c.setupConnection(setupCtx); err != nil {
c.log.Error().Err(err).Msg("error during setup connection")
}
})
c.log.Warn().Msg("Connection used in an unsafe way. !!!DON'T USE IN PRODUCTION!!!")
// Here we write directly and don't wait till the connection is "really" connected
- // Note: we can't use fireConnected here as it's guarded against c.driverContext.awaitSetupComplete
c.SetConnected(true)
return nil
}
if err := c.setupConnection(ctx); err != nil {
+ if disconnectErr := c.messageCodec.Disconnect(); disconnectErr != nil {
+ c.log.Debug().Err(disconnectErr).Msg("error disconnecting after failed setup")
+ }
return errors.Wrap(err, "error during setup connection")
}
return nil
@@ -153,278 +160,240 @@ func (c *Connection) Close() error {
ctx := context.TODO()
ctx, cancelFunc := utils.WithNamedTimeout(ctx, "connection close timeout", 5*time.Second)
defer cancelFunc()
+ if c.sessionState.connectionId != 0 {
+ c.log.Debug().Msg("Sending ForwardClose request")
+ forwardClose := readWriteModel.NewCipRRData(
+ c.sessionState.sessionHandle, uint32(readWriteModel.CIPStatus_Success), c.sessionState.senderContext, 0,
+ EmptyInterfaceHandle, 0,
+ []readWriteModel.TypeId{
+ readWriteModel.NewNullAddressItem(),
+ readWriteModel.NewUnConnectedDataItem(readWriteModel.NewCipConnectionManagerCloseRequest(
+ 2,
+ readWriteModel.NewLogicalSegment(readWriteModel.NewClassID(0, 6)),
+ readWriteModel.NewLogicalSegment(readWriteModel.NewInstanceID(0, 1)),
+ 0, 10, 14,
+ c.sessionState.connectionSerialNumber, 4919, 42,
+ c.sessionState.connectionPathSize, c.sessionState.routingAddress,
+ )),
+ },
+ )
+ // Many devices close the socket right after this - errors are expected and ignored.
+ if err := c.messageCodec.Send(ctx, "forward_close", forwardClose); err != nil {
+ c.log.Debug().Err(err).Msg("error sending forward close")
+ }
+ c.sessionState.connectionId = 0
+ }
c.log.Debug().Msg("Sending UnregisterSession EIP Packet")
- if err := c.messageCodec.SendRequest(ctx, "close_eip_disconnect_request", readWriteModel.NewEipDisconnectRequest(c.sessionState.sessionHandle, 0, []byte(DefaultSenderContext), 0), func(message spi.Message) bool {
- return true
- }, func(message spi.Message) error {
- return nil
- }, func(err error) error {
- return nil
- }); err != nil {
+ if err := c.messageCodec.Send(ctx, "unregister_session",
+ readWriteModel.NewEipDisconnectRequest(c.sessionState.sessionHandle, 0, []byte(DefaultSenderContext), 0),
+ ); err != nil {
c.log.Debug().Err(err).Msg("error sending unregister session request")
}
-
- //Unregister gets no response
- time.Sleep(100 * time.Millisecond) // Just to make sure it ge's out
+ // Unregister gets no response
+ time.Sleep(100 * time.Millisecond) // Just to make sure it gets out
if err := c.messageCodec.Disconnect(); err != nil {
c.log.Warn().Err(err).Msg("error disconnecting message codec")
}
c.log.Debug().
Uint32("sessionHandle", c.sessionState.sessionHandle).
- Msg("Unregistred Session %d")
+ Msg("Unregistered session")
return nil
}
func (c *Connection) setupConnection(ctx context.Context) error {
- if err := c.listServiceRequest(ctx); err != nil {
- return errors.Wrap(err, "error listing service request")
+ if err := c.listServices(ctx); err != nil {
+ return errors.Wrap(err, "error listing services")
}
-
- if err := c.connectRegisterSession(ctx); err != nil {
- return errors.Wrap(err, "error connect register session")
+ if err := c.registerSession(ctx); err != nil {
+ return errors.Wrap(err, "error registering session")
}
-
- if err := c.listAllAttributes(ctx); err != nil {
- return errors.Wrap(err, "error list all attributes")
+ if c.configuration.forceUnconnectedOperation {
+ c.log.Debug().Msg("Unconnected operation forced, skipping the capability probe")
+ c.SetConnected(true)
+ return nil
+ }
+ if err := c.probeAttributes(ctx); err != nil {
+ // Any probe failure (timeout, error status, malformed reply) means "device has
+ // no message router / connection manager": fall back to unconnected operation
+ // like plc4j instead of failing the connect.
+ c.log.Debug().Err(err).Msg("GetAttributeAll probe failed, falling back to unconnected mode")
+ c.sessionState.useMessageRouter = false
+ c.sessionState.useConnectionManager = false
}
-
if c.sessionState.useConnectionManager {
- // TODO: Continue here ....
- } else {
- c.SetConnected(true)
+ if err := c.openConnectionManager(ctx); err != nil {
+ // Deliberate deviation from plc4j (which fails the connect here): a device
+ // that advertises a connection manager but rejects the ForwardOpen is still
+ // usable through unconnected messaging.
+ c.log.Debug().Err(err).Msg("ForwardOpen failed, falling back to unconnected mode")
+ c.sessionState.useConnectionManager = false
+ }
}
+ c.SetConnected(true)
return nil
}
-func (c *Connection) listServiceRequest(ctx context.Context) error {
- c.log.Debug().Msg("Sending ListServices Request")
- listServicesResultChan := make(chan readWriteModel.ListServicesResponse, 1)
- listServicesResultErrorChan := make(chan error, 1)
- if err := c.messageCodec.SendRequest(ctx, "list_service_request", readWriteModel.NewListServicesRequest(
- EmptySessionHandle,
- uint32(readWriteModel.CIPStatus_Success),
- []byte(DefaultSenderContext),
- uint32(0),
- ), func(message spi.Message) bool {
- eipPacket, ok := message.(readWriteModel.EipPacket)
- if !ok {
- return false
- }
- eipPacketListServicesResponse := eipPacket.(readWriteModel.ListServicesResponse)
- return eipPacketListServicesResponse != nil
- }, func(message spi.Message) error {
- listServicesResponse := message.(readWriteModel.ListServicesResponse)
- serviceResponse := listServicesResponse.GetTypeIds()[0].(readWriteModel.ServicesResponse)
- if serviceResponse.GetSupportsCIPEncapsulation() {
- c.log.Debug().Msg("Device is capable of CIP over EIP encapsulation")
- }
- c.cipEncapsulationAvailable = serviceResponse.GetSupportsCIPEncapsulation()
- listServicesResultChan <- listServicesResponse
- return nil
- }, func(err error) error {
- // If this is a timeout, do a check if the connection requires a reconnection
- var timeoutError utils.TimeoutError
- if errors.As(err, &timeoutError) {
- c.log.Warn().Msg("Timeout during Connection establishing, closing channel...")
- c.Close()
- }
- listServicesResultErrorChan <- errors.Wrap(err, "got error processing request")
- return nil
- }); err != nil {
- return errors.Wrap(err, "Error during sending of EIP ListServices Request")
+func (c *Connection) listServices(ctx context.Context) error {
+ c.log.Debug().Msg("Sending ListServices request")
+ message, err := sendRequestAndWait(ctx, c.log, c.messageCodec, "list_services",
+ readWriteModel.NewListServicesRequest(
+ EmptySessionHandle, uint32(readWriteModel.CIPStatus_Success), []byte(DefaultSenderContext), 0,
+ ), func(message spi.Message) bool {
+ _, ok := message.(readWriteModel.ListServicesResponse)
+ return ok
+ })
+ if err != nil {
+ return err
}
-
- select {
- case <-ctx.Done():
- return errors.Wrap(ctx.Err(), "timeout")
- case err := <-listServicesResultErrorChan:
- return errors.Wrap(err, "Error receiving of ListServices response")
- case _ = <-listServicesResultChan:
- return nil
+ listServicesResponse := message.(readWriteModel.ListServicesResponse)
+ if len(listServicesResponse.GetTypeIds()) == 0 {
+ return errors.New("ListServices response contains no services")
+ }
+ servicesResponse, ok := listServicesResponse.GetTypeIds()[0].(readWriteModel.ServicesResponse)
+ if !ok {
+ return errors.Errorf("unexpected type id in ListServices response: %T", listServicesResponse.GetTypeIds()[0])
}
+ if !servicesResponse.GetSupportsCIPEncapsulation() {
+ return errors.New("device does not support CIP encapsulation")
+ }
+ c.cipEncapsulationAvailable = true
+ c.log.Debug().Msg("Device supports CIP over EIP encapsulation")
+ return nil
}
-func (c *Connection) connectRegisterSession(ctx context.Context) error {
- c.log.Debug().Msg("Sending EipConnectionRequest")
- connectionResponseChan := make(chan readWriteModel.EipConnectionResponse, 1)
- connectionResponseErrorChan := make(chan error, 1)
- if err := c.messageCodec.SendRequest(ctx, "connect_register_session", readWriteModel.NewEipConnectionRequest(
- EmptySessionHandle,
- uint32(readWriteModel.CIPStatus_Success),
- []byte(DefaultSenderContext),
- uint32(0),
- ), func(message spi.Message) bool {
- _, ok := message.(readWriteModel.EipPacket)
- return ok
- }, func(message spi.Message) error {
- eipPacket := message.(readWriteModel.EipPacket)
- connectionResponse := eipPacket.(readWriteModel.EipConnectionResponse)
- if connectionResponse != nil {
- if connectionResponse.GetStatus() == 0 {
- c.sessionState.sessionHandle = connectionResponse.GetSessionHandle()
- c.sessionState.senderContext = connectionResponse.GetSenderContext()
- c.log.Debug().
- Uint32("sessionHandle", c.sessionState.sessionHandle).
- Msg("Got assigned with Session")
- connectionResponseChan <- connectionResponse
- } else {
- c.log.Error().
- Uint32("status", connectionResponse.GetStatus()).
- Msg("Got unsuccessful status for connection request")
- connectionResponseErrorChan <- errors.New("got unsuccessful connection response")
- }
- } else {
- // TODO: This seems pretty hard-coded ... possibly find out if we can't simplify this.
- classSegment := readWriteModel.NewLogicalSegment(readWriteModel.NewClassID(0, 6))
- instanceSegment := readWriteModel.NewLogicalSegment(readWriteModel.NewClassID(0, 1))
- exchange := readWriteModel.NewUnConnectedDataItem(
- readWriteModel.NewCipConnectionManagerRequest(classSegment, instanceSegment, 0, 10,
- 14, 536870914, 33944, c.sessionState.connectionSerialNumber,
- 4919, 42, 3, 2101812,
- readWriteModel.NewNetworkConnectionParameters(4002, false, 2, 0, true),
- 2113537,
- readWriteModel.NewNetworkConnectionParameters(4002, false, 2, 0, true),
- readWriteModel.NewTransportType(true, 2, 3),
- c.sessionState.connectionPathSize, c.sessionState.routingAddress))
- typeIds := []readWriteModel.TypeId{readWriteModel.NewNullAddressItem(), exchange}
- eipWrapper := readWriteModel.NewCipRRData(
- c.sessionState.sessionHandle,
- uint32(readWriteModel.CIPStatus_Success),
- c.sessionState.senderContext,
- 0,
- c.sessionState.sessionHandle,
- 0,
- typeIds,
- )
- if err := c.messageCodec.SendRequest(ctx, "todo_what_is_this", eipWrapper, func(message spi.Message) bool {
- eipPacket := message.(readWriteModel.EipPacket)
- if eipPacket == nil {
- return false
- }
- cipRRData := eipPacket.(readWriteModel.CipRRData)
- return cipRRData != nil
- }, func(message spi.Message) error {
- cipRRData := message.(readWriteModel.CipRRData)
- if cipRRData.GetStatus() == 0 {
- unconnectedDataItem := cipRRData.GetTypeIds()[1].(readWriteModel.UnConnectedDataItem)
- connectionManagerResponse := unconnectedDataItem.GetService().(readWriteModel.CipConnectionManagerResponse)
- c.sessionState.connectionId = connectionManagerResponse.GetOtConnectionId()
- c.log.Debug().
- Uint32("connectionId", c.sessionState.connectionId).
- Msg("Got assigned with connection if")
- connectionResponseChan <- connectionResponse
- } else {
- connectionResponseErrorChan <- fmt.Errorf("got status code while opening Connection manager: %d", cipRRData.GetStatus())
- }
- return nil
- }, func(err error) error {
- // If this is a timeout, do a check if the connection requires a reconnection
- var timeoutError utils.TimeoutError
- if errors.As(err, &timeoutError) {
- c.log.Warn().Msg("Timeout during Connection establishing, closing channel...")
- c.Close()
- }
- connectionResponseErrorChan <- errors.Wrap(err, "got error processing request")
- return nil
- }); err != nil {
- connectionResponseErrorChan <- errors.Wrap(err, "Error during sending of EIP ListServices Request")
- }
- }
- return nil
- }, func(err error) error {
- // If this is a timeout, do a check if the connection requires a reconnection
- var timeoutError utils.TimeoutError
- if errors.As(err, &timeoutError) {
- c.log.Warn().Msg("Timeout during Connection establishing, closing channel...")
- if err := c.Close(); err != nil {
- c.log.Debug().Err(err).Msg("Error closing connection")
- }
- }
- connectionResponseErrorChan <- errors.Wrap(err, "got error processing request")
- return nil
- }); err != nil {
- return errors.Wrap(err, "Error during sending of EIP ListServices Request")
- }
- select {
- case <-ctx.Done():
- return errors.Wrap(ctx.Err(), "timeout")
- case err := <-connectionResponseErrorChan:
- return errors.Wrap(err, "Error receiving of ListServices response")
- case _ = <-connectionResponseChan:
+func (c *Connection) registerSession(ctx context.Context) error {
+ c.log.Debug().Msg("Sending RegisterSession request")
+ message, err := sendRequestAndWait(ctx, c.log, c.messageCodec, "register_session",
+ readWriteModel.NewEipConnectionRequest(
+ EmptySessionHandle, uint32(readWriteModel.CIPStatus_Success), []byte(DefaultSenderContext), 0,
+ ), func(message spi.Message) bool {
+ _, ok := message.(readWriteModel.EipPacket)
+ return ok
+ })
+ if err != nil {
+ return err
+ }
+ connectionResponse, ok := message.(readWriteModel.EipConnectionResponse)
+ if !ok {
+ // Some devices skip ahead to the connection manager - proceed without a
+ // registered session, like plc4j.
+ c.log.Debug().Type("responseType", message).Msg("Device skipped session registration")
return nil
}
+ if connectionResponse.GetStatus() != uint32(readWriteModel.CIPStatus_Success) {
+ return errors.Errorf("got status code while registering session [%d]", connectionResponse.GetStatus())
+ }
+ c.sessionState.sessionHandle = connectionResponse.GetSessionHandle()
+ c.sessionState.senderContext = connectionResponse.GetSenderContext()
+ c.log.Debug().Uint32("sessionHandle", c.sessionState.sessionHandle).Msg("Got assigned with session handle")
+ return nil
}
-func (c *Connection) listAllAttributes(ctx context.Context) error {
- c.log.Debug().Msg("Sending ListAllAttributes Request")
- listAllAttributesResponseChan := make(chan readWriteModel.GetAttributeAllResponse, 1)
- listAllAttributesErrorChan := make(chan error, 1)
- classSegment := readWriteModel.NewLogicalSegment(readWriteModel.NewClassID(uint8(0), uint8(2)))
- instanceSegment := readWriteModel.NewLogicalSegment(readWriteModel.NewInstanceID(uint8(0), uint8(1)))
- if err := c.messageCodec.SendRequest(ctx, "list_all_attributes", readWriteModel.NewCipRRData(
- c.sessionState.sessionHandle,
- uint32(readWriteModel.CIPStatus_Success),
- c.sessionState.senderContext,
- 0,
- EmptyInterfaceHandle,
- 0,
+func (c *Connection) probeAttributes(ctx context.Context) error {
+ c.log.Debug().Msg("Sending GetAttributeAll probe")
+ probe := readWriteModel.NewCipRRData(
+ c.sessionState.sessionHandle, uint32(readWriteModel.CIPStatus_Success), c.sessionState.senderContext, 0,
+ EmptyInterfaceHandle, 0,
[]readWriteModel.TypeId{
readWriteModel.NewNullAddressItem(),
- readWriteModel.NewUnConnectedDataItem(
- readWriteModel.NewGetAttributeAllRequest(classSegment, instanceSegment),
- ),
+ readWriteModel.NewUnConnectedDataItem(readWriteModel.NewGetAttributeAllRequest(
+ readWriteModel.NewLogicalSegment(readWriteModel.NewClassID(0, 2)),
+ readWriteModel.NewLogicalSegment(readWriteModel.NewInstanceID(0, 1)),
+ )),
},
- ), func(message spi.Message) bool {
- eipPacket := message.(readWriteModel.CipRRData)
- return eipPacket != nil
- }, func(message spi.Message) error {
- cipRrData := message.(readWriteModel.CipRRData)
- if cipRrData.GetStatus() == uint32(readWriteModel.CIPStatus_Success) {
- dataItem := cipRrData.GetTypeIds()[1].(readWriteModel.UnConnectedDataItem)
- response := dataItem.GetService().(readWriteModel.GetAttributeAllResponse)
- if response.GetStatus() != uint8(readWriteModel.CIPStatus_Success) {
- // TODO: Return an error ...
- } else if response.GetAttributes() != nil {
- for _, classId := range response.GetAttributes().GetClassId() {
- if curCipClassId, ok := readWriteModel.CIPClassIDByValue(classId); ok {
- switch curCipClassId {
- case readWriteModel.CIPClassID_MessageRouter:
- c.sessionState.useMessageRouter = true
- case readWriteModel.CIPClassID_ConnectionManager:
- c.sessionState.useConnectionManager = true
- }
- }
+ )
+ message, err := sendRequestAndWait(ctx, c.log, c.messageCodec, "get_attribute_all", probe,
+ func(message spi.Message) bool {
+ _, ok := message.(readWriteModel.CipRRData)
+ return ok
+ })
+ if err != nil {
+ return err
+ }
+ cipRRData := message.(readWriteModel.CipRRData)
+ if cipRRData.GetStatus() != uint32(readWriteModel.CIPStatus_Success) {
+ return errors.Errorf("got status code on GetAttributeAll probe [%d]", cipRRData.GetStatus())
+ }
+ if len(cipRRData.GetTypeIds()) < 2 {
+ return errors.New("GetAttributeAll probe response contains no data item")
+ }
+ dataItem, ok := cipRRData.GetTypeIds()[1].(readWriteModel.UnConnectedDataItem)
+ if !ok {
+ return errors.Errorf("unexpected type id in GetAttributeAll probe response: %T", cipRRData.GetTypeIds()[1])
+ }
+ response, ok := dataItem.GetService().(readWriteModel.GetAttributeAllResponse)
+ if !ok {
+ return errors.Errorf("unexpected service in GetAttributeAll probe response: %T", dataItem.GetService())
+ }
+ if response.GetStatus() != uint8(readWriteModel.CIPStatus_Success) {
+ return errors.Errorf("got status code on GetAttributeAll response [%d]", response.GetStatus())
+ }
+ if attributes := response.GetAttributes(); attributes != nil {
+ for _, classId := range attributes.GetClassId() {
+ if cipClassId, ok := readWriteModel.CIPClassIDByValue(classId); ok {
+ switch cipClassId {
+ case readWriteModel.CIPClassID_MessageRouter:
+ c.sessionState.useMessageRouter = true
+ case readWriteModel.CIPClassID_ConnectionManager:
+ c.sessionState.useConnectionManager = true
+ default:
}
}
- c.log.Debug().
- Bool("useMessageRouter", c.sessionState.useMessageRouter).
- Bool("useConnectionManager", c.sessionState.useConnectionManager).
- Msg("Connection using message router, using connection manager")
- listAllAttributesResponseChan <- response
- }
- return nil
- }, func(err error) error {
- // If this is a timeout, do a check if the connection requires a reconnection
- if errors.Is(err, utils.TimeoutError{}) {
- c.log.Warn().Msg("Timeout during Connection establishing, closing channel...")
- if err := c.Close(); err != nil {
- c.log.Debug().Err(err).Msg("Error closing connection")
- }
}
- listAllAttributesErrorChan <- errors.Wrap(err, "got error processing request")
- return nil
- }); err != nil {
- return errors.Wrap(err, "Error during sending of EIP ListServices Request")
}
+ c.log.Debug().
+ Bool("useMessageRouter", c.sessionState.useMessageRouter).
+ Bool("useConnectionManager", c.sessionState.useConnectionManager).
+ Msg("Probed device capabilities")
+ return nil
+}
- select {
- case <-ctx.Done():
- return errors.Wrap(ctx.Err(), "timeout")
- case err := <-listAllAttributesErrorChan:
- return errors.Wrap(err, "Error receiving of ListServices response")
- case _ = <-listAllAttributesResponseChan:
- return nil
+func (c *Connection) openConnectionManager(ctx context.Context) error {
+ c.log.Debug().Msg("Sending ForwardOpen request")
+ forwardOpen := readWriteModel.NewCipRRData(
+ c.sessionState.sessionHandle, uint32(readWriteModel.CIPStatus_Success), c.sessionState.senderContext, 0,
+ EmptyInterfaceHandle, 0,
+ []readWriteModel.TypeId{
+ readWriteModel.NewNullAddressItem(),
+ readWriteModel.NewUnConnectedDataItem(readWriteModel.NewCipConnectionManagerRequest(
+ readWriteModel.NewLogicalSegment(readWriteModel.NewClassID(0, 6)),
+ readWriteModel.NewLogicalSegment(readWriteModel.NewInstanceID(0, 1)),
+ 0, 10, 14, 536870914, 33944,
+ c.sessionState.connectionSerialNumber, 4919, 42, 3, 2101812,
+ readWriteModel.NewNetworkConnectionParameters(4002, false, 2, 0, true),
+ 2113537,
+ readWriteModel.NewNetworkConnectionParameters(4002, false, 2, 0, true),
+ readWriteModel.NewTransportType(true, 2, 3),
+ c.sessionState.connectionPathSize, c.sessionState.routingAddress,
+ )),
+ },
+ )
+ message, err := sendRequestAndWait(ctx, c.log, c.messageCodec, "forward_open", forwardOpen,
+ func(message spi.Message) bool {
+ _, ok := message.(readWriteModel.CipRRData)
+ return ok
+ })
+ if err != nil {
+ return err
+ }
+ cipRRData := message.(readWriteModel.CipRRData)
+ if cipRRData.GetStatus() != uint32(readWriteModel.CIPStatus_Success) {
+ return errors.Errorf("got status code while opening connection manager [%d]", cipRRData.GetStatus())
}
+ if len(cipRRData.GetTypeIds()) < 2 {
+ return errors.New("ForwardOpen response contains no data item")
+ }
+ dataItem, ok := cipRRData.GetTypeIds()[1].(readWriteModel.UnConnectedDataItem)
+ if !ok {
+ return errors.Errorf("unexpected type id in ForwardOpen response: %T", cipRRData.GetTypeIds()[1])
+ }
+ connectionManagerResponse, ok := dataItem.GetService().(readWriteModel.CipConnectionManagerResponse)
+ if !ok {
+ return errors.Errorf("unexpected service in ForwardOpen response: %T", dataItem.GetService())
+ }
+ c.sessionState.connectionId = connectionManagerResponse.GetOtConnectionId()
+ c.log.Debug().Uint32("connectionId", c.sessionState.connectionId).Msg("Got assigned with connection id")
+ return nil
}
func (c *Connection) GetMetadata() apiModel.PlcConnectionMetadata {
diff --git a/plc4go/internal/eip/MessageCodec.go b/plc4go/internal/eip/MessageCodec.go
index ae6d17f41e..98ecad5fbb 100644
--- a/plc4go/internal/eip/MessageCodec.go
+++ b/plc4go/internal/eip/MessageCodec.go
@@ -87,6 +87,22 @@ func (m *MessageCodec) Send(ctx context.Context, interactionInfo string, message
func (m *MessageCodec) Receive(ctx context.Context) (spi.Message, error) {
// We need at least 6 bytes in order to know how big the packet is in total
transportInstance := m.GetTransportInstance()
+ // Pull data from the transport until at least the 4-byte header is buffered.
+ // Some transports (e.g. the test transport) only surface queued data through fills,
+ // so checking the buffer without filling first would starve the receive worker.
+ if err := transportInstance.FillBuffer(ctx, func(pos uint, currentByte byte, reader transports.ExtendedReader) bool {
+ numBytesAvailable, err := transportInstance.GetNumBytesAvailableInBuffer()
+ if err != nil {
+ return false
+ }
+ return numBytesAvailable < 4
+ }); err != nil {
+ if transportError, ok := transports.AsTransportError(err); ok && transportError.Kind() == transports.TransportErrorFatal {
+ return nil, err
+ }
+ // Fall through on non-fatal errors, we might have enough data buffered already.
+ m.log.Trace().Err(err).Msg("Error filling buffer, continuing with what's available")
+ }
if num, err := transportInstance.GetNumBytesAvailableInBuffer(); (err == nil) && (num >= 4) {
m.log.Debug().Uint32("num", num).Msg("we got num readable bytes")
data, err := transportInstance.PeekReadableBytes(ctx, 4)
diff --git a/plc4go/internal/eip/Requests.go b/plc4go/internal/eip/Requests.go
new file mode 100644
index 0000000000..870b9e2b3b
--- /dev/null
+++ b/plc4go/internal/eip/Requests.go
@@ -0,0 +1,102 @@
+/*
+ * 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 eip
+
+import (
+ "context"
+
+ "github.com/rs/zerolog"
+
+ "github.com/apache/plc4x/plc4go/spi"
+ "github.com/apache/plc4x/plc4go/spi/errors"
+ "github.com/apache/plc4x/plc4go/spi/transactions"
+)
+
+// sendRequestAndWait sends one request through the codec and blocks until the
+// accepted response, an error, or ctx cancellation. Every outcome resolves -
+// no code path may leave the caller waiting forever.
+func sendRequestAndWait(
+ ctx context.Context,
+ log zerolog.Logger,
+ codec spi.MessageCodec,
+ name string,
+ request spi.Message,
+ accepts func(spi.Message) bool,
+) (spi.Message, error) {
+ messageChan := make(chan spi.Message, 1)
+ errChan := make(chan error, 1)
+ if err := codec.SendRequest(ctx, name, request, accepts, func(message spi.Message) error {
+ messageChan <- message
+ return nil
+ }, func(err error) error {
+ errChan <- err
+ return nil
+ }); err != nil {
+ return nil, errors.Wrapf(err, "error sending %s request", name)
+ }
+ select {
+ case <-ctx.Done():
+ return nil, errors.Wrapf(ctx.Err(), "%s aborted", name)
+ case err := <-errChan:
+ return nil, errors.Wrapf(err, "error waiting for %s response", name)
+ case message := <-messageChan:
+ return message, nil
+ }
+}
+
+// sendTransactedRequestAndWait is sendRequestAndWait routed through the
+// request transaction manager, for the read/write paths.
+func sendTransactedRequestAndWait(
+ ctx context.Context,
+ log zerolog.Logger,
+ codec spi.MessageCodec,
+ tm transactions.RequestTransactionManager,
+ name string,
+ request spi.Message,
+ accepts func(spi.Message) bool,
+) (spi.Message, error) {
+ messageChan := make(chan spi.Message, 1)
+ errChan := make(chan error, 1)
+ transaction := tm.StartTransaction(name)
+ transaction.Submit(name+"Operation", func(transactionContext context.Context, transaction transactions.RequestTransaction) {
+ ctx, cancel := context.WithCancel(ctx)
+ context.AfterFunc(transactionContext, cancel)
+ if err := codec.SendRequest(ctx, name, request, accepts, func(message spi.Message) error {
+ messageChan <- message
+ return transaction.EndRequest()
+ }, func(err error) error {
+ errChan <- err
+ return transaction.EndRequest()
+ }); err != nil {
+ errChan <- errors.Wrapf(err, "error sending %s request", name)
+ if failErr := transaction.FailRequest(err); failErr != nil {
+ log.Debug().Err(failErr).Msg("error failing request")
+ }
+ }
+ })
+ select {
+ case <-ctx.Done():
+ return nil, errors.Wrapf(ctx.Err(), "%s aborted", name)
+ case err := <-errChan:
+ return nil, errors.Wrapf(err, "error waiting for %s response", name)
+ case message := <-messageChan:
+ return message, nil
+ }
+}
diff --git a/plc4go/internal/eip/Tag.go b/plc4go/internal/eip/Tag.go
index 112fa3e3b9..6b14bf4a6f 100644
--- a/plc4go/internal/eip/Tag.go
+++ b/plc4go/internal/eip/Tag.go
@@ -89,6 +89,14 @@ func (m plcTag) Serialize() ([]byte, error) {
}
func (m plcTag) SerializeWithWriteBuffer(ctx context.Context, wb utils.WriteBuffer) error {
+ // The driver testsuite golden expects tags wrapped as PlcTagItem/tag/EipTag
+ // (mirroring plc4j's PlcTagItem), rather than a bare EipTag.
+ if err := wb.PushContext("PlcTagItem"); err != nil {
+ return err
+ }
+ if err := wb.PushContext("tag"); err != nil {
+ return err
+ }
if err := wb.PushContext("EipTag"); err != nil {
return err
}
@@ -110,6 +118,12 @@ func (m plcTag) SerializeWithWriteBuffer(ctx context.Context, wb utils.WriteBuff
if err := wb.PopContext("EipTag"); err != nil {
return err
}
+ if err := wb.PopContext("tag"); err != nil {
+ return err
+ }
+ if err := wb.PopContext("PlcTagItem"); err != nil {
+ return err
+ }
return nil
}
diff --git a/plc4go/protocols/eip/readwrite/model/AnsiExtendedSymbolSegment.go b/plc4go/protocols/eip/readwrite/model/AnsiExtendedSymbolSegment.go
index 0b036bb4eb..c4dce71a83 100644
--- a/plc4go/protocols/eip/readwrite/model/AnsiExtendedSymbolSegment.go
+++ b/plc4go/protocols/eip/readwrite/model/AnsiExtendedSymbolSegment.go
@@ -26,6 +26,7 @@ import (
"github.com/rs/zerolog"
+ "github.com/apache/plc4x/plc4go/spi/codegen"
. "github.com/apache/plc4x/plc4go/spi/codegen/fields"
. "github.com/apache/plc4x/plc4go/spi/codegen/io"
"github.com/apache/plc4x/plc4go/spi/errors"
@@ -306,7 +307,7 @@ func (m *_AnsiExtendedSymbolSegment) SerializeWithWriteBuffer(ctx context.Contex
return errors.Wrap(err, "Error serializing 'dataSize' field")
}
- if err := WriteSimpleField[string](ctx, "symbol", m.GetSymbol(), WriteString(writeBuffer, int32(int32(uint8(len(m.GetSymbol())))*int32(int32(8))))); err != nil {
+ if err := WriteSimpleField[string](ctx, "symbol", m.GetSymbol(), WriteString(writeBuffer, int32(int32(uint8(len(m.GetSymbol())))*int32(int32(8)))), codegen.WithEncoding("UTF8")); err != nil {
return errors.Wrap(err, "Error serializing 'symbol' field")
}
diff --git a/plc4go/tests/drivers/tests/eip_driver_test.go b/plc4go/tests/drivers/tests/eip_driver_test.go
index f8f1f83493..663912f854 100644
--- a/plc4go/tests/drivers/tests/eip_driver_test.go
+++ b/plc4go/tests/drivers/tests/eip_driver_test.go
@@ -28,7 +28,6 @@ import (
)
func TestEIPDriver(t *testing.T) {
- t.Skip("temporary skip for really strange bug") // TODO: unskip asap
optionsForTesting := testutils.EnrichOptionsWithOptionsForTesting(t)
testutils.RunDriverTestsuite(
t,