You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by sr...@apache.org on 2023/07/31 18:12:06 UTC
[plc4x] branch develop updated (c9efef33bc -> 5c014a0536)
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a change to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
from c9efef33bc refactor(plc4j/opcua): slight cleanup and qc fixing
new dcc5edba2c feat(plc4go): implement unsubscription requests
new 5c014a0536 feat(plc4go/opcua): implement unsubscription
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
plc4go/internal/opcua/Connection.go | 1 -
plc4go/internal/opcua/MessageCodec.go | 2 +-
plc4go/internal/opcua/Subscriber.go | 8 +--
plc4go/internal/opcua/SubscriptionHandle.go | 73 ++++++++++++++++++++++
plc4go/pkg/api/model/plc_unsubscription.go | 10 +--
.../spi/model/DefaultPlcUnsubscriptionRequest.go | 36 ++++++++---
...ultPlcUnsubscriptionRequestBuilder_plc4xgen.go} | 22 +++----
.../DefaultPlcUnsubscriptionRequest_plc4xgen.go | 45 ++++++++++++-
.../spi/model/DefaultPlcUnsubscriptionResponse.go | 12 ++--
.../DefaultPlcUnsubscriptionResponse_plc4xgen.go | 2 +-
plc4go/spi/model/render_test.go | 4 +-
11 files changed, 175 insertions(+), 40 deletions(-)
copy plc4go/spi/model/{DefaultPlcBrowseResponseItem_plc4xgen.go => DefaultPlcUnsubscriptionRequestBuilder_plc4xgen.go} (71%)
[plc4x] 02/02: feat(plc4go/opcua): implement unsubscription
Posted by sr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 5c014a0536e7e3e37fbf3eecc1eb3c260a3b9883
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Mon Jul 31 20:11:59 2023 +0200
feat(plc4go/opcua): implement unsubscription
---
plc4go/internal/opcua/Connection.go | 1 -
plc4go/internal/opcua/MessageCodec.go | 2 +-
plc4go/internal/opcua/Subscriber.go | 8 ++--
plc4go/internal/opcua/SubscriptionHandle.go | 73 +++++++++++++++++++++++++++++
4 files changed, 77 insertions(+), 7 deletions(-)
diff --git a/plc4go/internal/opcua/Connection.go b/plc4go/internal/opcua/Connection.go
index f96ad70a94..b5ea89a257 100644
--- a/plc4go/internal/opcua/Connection.go
+++ b/plc4go/internal/opcua/Connection.go
@@ -191,7 +191,6 @@ func (c *Connection) addSubscriber(subscriber *Subscriber) {
}
func (c *Connection) setupConnection(ctx context.Context, ch chan plc4go.PlcConnectionConnectResult) {
- // TODO: what to do?
c.log.Trace().Msg("Connection setup done")
c.fireConnected(ch)
c.log.Trace().Msg("Connect fired")
diff --git a/plc4go/internal/opcua/MessageCodec.go b/plc4go/internal/opcua/MessageCodec.go
index 73675fecaa..1114303dd2 100644
--- a/plc4go/internal/opcua/MessageCodec.go
+++ b/plc4go/internal/opcua/MessageCodec.go
@@ -163,7 +163,7 @@ func (m *MessageCodec) Receive() (spi.Message, error) {
if err != nil {
return nil, errors.Wrapf(err, "could not read %d bytes", readBytes)
}
- ctxForModel := options.GetLoggerContextForModel(context.TODO(), m.log, options.WithPassLoggerToModel(m.passLogToModel))
+ ctxForModel := options.GetLoggerContextForModel(context.Background(), m.log, options.WithPassLoggerToModel(m.passLogToModel))
messagePdu, err := readWriteModel.MessagePDUParse(ctxForModel, readBytes, true)
if err != nil {
return nil, errors.New("Could not parse pdu")
diff --git a/plc4go/internal/opcua/Subscriber.go b/plc4go/internal/opcua/Subscriber.go
index cd815756f5..f485994f77 100644
--- a/plc4go/internal/opcua/Subscriber.go
+++ b/plc4go/internal/opcua/Subscriber.go
@@ -202,14 +202,12 @@ func (s *Subscriber) onDisconnect(codec *MessageCodec) {
}
func (s *Subscriber) Unsubscribe(ctx context.Context, unsubscriptionRequest apiModel.PlcUnsubscriptionRequest) <-chan apiModel.PlcUnsubscriptionRequestResult {
- // TODO: handle context
result := make(chan apiModel.PlcUnsubscriptionRequestResult, 1)
result <- spiModel.NewDefaultPlcUnsubscriptionRequestResult(unsubscriptionRequest, nil, errors.New("Not Implemented"))
- // TODO: As soon as we establish a connection, we start getting data...
- // subscriptions are more a internal handling of which values to pass where.
- _ = ctx
- _ = unsubscriptionRequest
+ for _, handle := range unsubscriptionRequest.(*spiModel.DefaultPlcUnsubscriptionRequest).GetSubscriptionHandles() {
+ handle.(*SubscriptionHandle).stopSubscriber()
+ }
return result
}
diff --git a/plc4go/internal/opcua/SubscriptionHandle.go b/plc4go/internal/opcua/SubscriptionHandle.go
index 393bf8dc6f..15a601267e 100644
--- a/plc4go/internal/opcua/SubscriptionHandle.go
+++ b/plc4go/internal/opcua/SubscriptionHandle.go
@@ -363,8 +363,81 @@ func (h *SubscriptionHandle) startSubscriber() {
}()
}
+// stopSubscriber stops the subscriber either on disconnect or on error
func (h *SubscriptionHandle) stopSubscriber() {
+ h.destroy.Store(true)
+ requestHandle := h.channel.getRequestHandle()
+
+ requestHeader := readWriteModel.NewRequestHeader(
+ h.channel.getAuthenticationToken(),
+ h.channel.getCurrentDateTime(),
+ requestHandle,
+ 0,
+ NULL_STRING,
+ uint32(h.revisedCycleTime*10),
+ NULL_EXTENSION_OBJECT,
+ )
+
+ subscriptions := []uint32{h.subscriptionId}
+ deleteSubscriptionrequest := readWriteModel.NewDeleteSubscriptionsRequest(requestHeader,
+ 1,
+ subscriptions,
+ )
+
+ identifier, err := strconv.ParseUint(deleteSubscriptionrequest.GetIdentifier(), 10, 16)
+ if err != nil {
+ h.log.Error().Err(err).Msg("error parsing identifier")
+ return
+ }
+ extExpandedNodeId := readWriteModel.NewExpandedNodeId(false, //Namespace Uri Specified
+ false, //Server Index Specified
+ readWriteModel.NewNodeIdFourByte(0, uint16(identifier)),
+ nil,
+ nil,
+ )
+
+ extObject := readWriteModel.NewExtensionObject(
+ extExpandedNodeId,
+ nil,
+ deleteSubscriptionrequest,
+ false,
+ )
+
+ ctx := context.Background()
+
+ buffer := utils.NewWriteBufferByteBased(utils.WithByteOrderForByteBasedBuffer(binary.LittleEndian))
+ if err = extObject.SerializeWithWriteBuffer(ctx, buffer); err != nil {
+ h.log.Error().Err(err).Msg("Unable to serialise the ReadRequest")
+ return
+ }
+
+ consumer := func(opcuaResponse []byte) {
+ var responseMessage readWriteModel.DeleteSubscriptionsResponse
+ unknownExtensionObject, err := readWriteModel.ExtensionObjectParseWithBuffer(ctx, utils.NewReadBufferByteBased(opcuaResponse, utils.WithByteOrderForReadBufferByteBased(binary.LittleEndian)), false)
+ if err != nil {
+ h.log.Error().Err(err).Msg("Unable to parse the returned Subscription response")
+ h.plcSubscriber.onDisconnect(h.messageCodec)
+ return
+ }
+ switch unknownExtensionObject := unknownExtensionObject.(type) {
+ case readWriteModel.DeleteSubscriptionsResponse:
+ responseMessage = unknownExtensionObject
+ case readWriteModel.ServiceFault:
+ serviceFault := unknownExtensionObject
+ header := serviceFault.GetResponseHeader().(readWriteModel.ResponseHeader)
+ h.log.Debug().Msgf("Subscription ServiceFault returned from server with error code, '%s', ignoring as it is probably just a result of a Delete Subscription Request", header.GetServiceResult())
+ return
+ }
+ h.log.Debug().Msgf("Received response\n%s", responseMessage)
+ }
+
+ errorDispatcher := func(err error) {
+ h.log.Error().Err(err).Msg("error received")
+ h.plcSubscriber.onDisconnect(h.messageCodec)
+ }
+
+ h.messageCodec.channel.submit(ctx, h.messageCodec, errorDispatcher, consumer, buffer)
}
/**
[plc4x] 01/02: feat(plc4go): implement unsubscription requests
Posted by sr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit dcc5edba2c96e361a5811a879fffc82ad76a5876
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Mon Jul 31 20:11:37 2023 +0200
feat(plc4go): implement unsubscription requests
---
plc4go/pkg/api/model/plc_unsubscription.go | 10 +++--
.../spi/model/DefaultPlcUnsubscriptionRequest.go | 36 +++++++++++++----
...ultPlcUnsubscriptionRequestBuilder_plc4xgen.go} | 38 ++++++++++++++----
.../DefaultPlcUnsubscriptionRequest_plc4xgen.go | 45 +++++++++++++++++++++-
.../spi/model/DefaultPlcUnsubscriptionResponse.go | 12 ++++--
.../DefaultPlcUnsubscriptionResponse_plc4xgen.go | 2 +-
plc4go/spi/model/render_test.go | 4 +-
7 files changed, 120 insertions(+), 27 deletions(-)
diff --git a/plc4go/pkg/api/model/plc_unsubscription.go b/plc4go/pkg/api/model/plc_unsubscription.go
index 13a337769f..d95dc65088 100644
--- a/plc4go/pkg/api/model/plc_unsubscription.go
+++ b/plc4go/pkg/api/model/plc_unsubscription.go
@@ -21,16 +21,17 @@ package model
import "context"
-type PlcUnsubscriptionRequestBuilder interface {
- // TODO: Implement
-}
-
type PlcUnsubscriptionRequestResult interface {
GetRequest() PlcUnsubscriptionRequest
GetResponse() PlcUnsubscriptionResponse
GetErr() error
}
+type PlcUnsubscriptionRequestBuilder interface {
+ AddHandles(PlcSubscriptionHandle ...PlcSubscriptionHandle)
+ Build() (PlcUnsubscriptionRequest, error)
+}
+
type PlcUnsubscriptionRequest interface {
Execute() <-chan PlcUnsubscriptionRequestResult
ExecuteWithContext(ctx context.Context) <-chan PlcUnsubscriptionRequestResult
@@ -38,4 +39,5 @@ type PlcUnsubscriptionRequest interface {
}
type PlcUnsubscriptionResponse interface {
+ GetRequest() PlcUnsubscriptionRequest
}
diff --git a/plc4go/spi/model/DefaultPlcUnsubscriptionRequest.go b/plc4go/spi/model/DefaultPlcUnsubscriptionRequest.go
index 3ea2b25aed..e6f2f1f534 100644
--- a/plc4go/spi/model/DefaultPlcUnsubscriptionRequest.go
+++ b/plc4go/spi/model/DefaultPlcUnsubscriptionRequest.go
@@ -21,30 +21,50 @@ package model
import (
"context"
+ "github.com/apache/plc4x/plc4go/spi"
apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
)
+var _ apiModel.PlcUnsubscriptionRequestBuilder = &DefaultPlcUnsubscriptionRequestBuilder{}
+
+//go:generate go run ../../tools/plc4xgenerator/gen.go -type=DefaultPlcUnsubscriptionRequestBuilder
+type DefaultPlcUnsubscriptionRequestBuilder struct {
+ subscriptionHandles []apiModel.PlcSubscriptionHandle
+}
+
+func (d *DefaultPlcUnsubscriptionRequestBuilder) AddHandles(subscriptionHandles ...apiModel.PlcSubscriptionHandle) {
+ subscriptionHandles = append(subscriptionHandles, subscriptionHandles...)
+}
+
+func (d *DefaultPlcUnsubscriptionRequestBuilder) Build() (apiModel.PlcUnsubscriptionRequest, error) {
+ return NewDefaultPlcUnsubscriptionRequest(d.subscriptionHandles), nil
+}
+
var _ apiModel.PlcUnsubscriptionRequest = &DefaultPlcUnsubscriptionRequest{}
//go:generate go run ../../tools/plc4xgenerator/gen.go -type=DefaultPlcUnsubscriptionRequest
type DefaultPlcUnsubscriptionRequest struct {
- // TODO: implement
- implementMe string
+ subscriber spi.PlcSubscriber
+ subscriptionHandles []apiModel.PlcSubscriptionHandle
}
-func NewDefaultPlcUnsubscriptionRequest() *DefaultPlcUnsubscriptionRequest {
- return &DefaultPlcUnsubscriptionRequest{}
+func NewDefaultPlcUnsubscriptionRequest(subscriptionHandles []apiModel.PlcSubscriptionHandle) *DefaultPlcUnsubscriptionRequest {
+ return &DefaultPlcUnsubscriptionRequest{
+ subscriptionHandles: subscriptionHandles,
+ }
}
func (d *DefaultPlcUnsubscriptionRequest) Execute() <-chan apiModel.PlcUnsubscriptionRequestResult {
- //TODO implement me
- panic("implement me")
+ return d.ExecuteWithContext(context.Background())
}
func (d *DefaultPlcUnsubscriptionRequest) ExecuteWithContext(ctx context.Context) <-chan apiModel.PlcUnsubscriptionRequestResult {
- //TODO implement me
- panic("implement me")
+ return d.subscriber.Unsubscribe(ctx, d)
+}
+
+func (d *DefaultPlcUnsubscriptionRequest) GetSubscriptionHandles() []apiModel.PlcSubscriptionHandle {
+ return d.subscriptionHandles
}
func (d *DefaultPlcUnsubscriptionRequest) IsAPlcMessage() bool {
diff --git a/plc4go/spi/model/DefaultPlcUnsubscriptionResponse_plc4xgen.go b/plc4go/spi/model/DefaultPlcUnsubscriptionRequestBuilder_plc4xgen.go
similarity index 53%
copy from plc4go/spi/model/DefaultPlcUnsubscriptionResponse_plc4xgen.go
copy to plc4go/spi/model/DefaultPlcUnsubscriptionRequestBuilder_plc4xgen.go
index ddb8dc6798..ab23d48365 100644
--- a/plc4go/spi/model/DefaultPlcUnsubscriptionResponse_plc4xgen.go
+++ b/plc4go/spi/model/DefaultPlcUnsubscriptionRequestBuilder_plc4xgen.go
@@ -17,7 +17,7 @@
* under the License.
*/
-// Code generated by "plc4xgenerator -type=DefaultPlcUnsubscriptionResponse"; DO NOT EDIT.
+// Code generated by "plc4xgenerator -type=DefaultPlcUnsubscriptionRequestBuilder"; DO NOT EDIT.
package model
@@ -30,7 +30,7 @@ import (
var _ = fmt.Printf
-func (d *DefaultPlcUnsubscriptionResponse) Serialize() ([]byte, error) {
+func (d *DefaultPlcUnsubscriptionRequestBuilder) Serialize() ([]byte, error) {
wb := utils.NewWriteBufferByteBased(utils.WithByteOrderForByteBasedBuffer(binary.BigEndian))
if err := d.SerializeWithWriteBuffer(context.Background(), wb); err != nil {
return nil, err
@@ -38,21 +38,45 @@ func (d *DefaultPlcUnsubscriptionResponse) Serialize() ([]byte, error) {
return wb.GetBytes(), nil
}
-func (d *DefaultPlcUnsubscriptionResponse) SerializeWithWriteBuffer(ctx context.Context, writeBuffer utils.WriteBuffer) error {
- if err := writeBuffer.PushContext("PlcUnsubscriptionResponse"); err != nil {
+func (d *DefaultPlcUnsubscriptionRequestBuilder) SerializeWithWriteBuffer(ctx context.Context, writeBuffer utils.WriteBuffer) error {
+ if err := writeBuffer.PushContext("PlcUnsubscriptionRequestBuilder"); err != nil {
return err
}
+ if err := writeBuffer.PushContext("subscriptionHandles", utils.WithRenderAsList(true)); err != nil {
+ return err
+ }
+ for _, elem := range d.subscriptionHandles {
+ var elem any = elem
- if err := writeBuffer.WriteString("implementMe", uint32(len(d.implementMe)*8), "UTF-8", d.implementMe); err != nil {
+ if elem != nil {
+ if serializableField, ok := elem.(utils.Serializable); ok {
+ if err := writeBuffer.PushContext("value"); err != nil {
+ return err
+ }
+ if err := serializableField.SerializeWithWriteBuffer(ctx, writeBuffer); err != nil {
+ return err
+ }
+ if err := writeBuffer.PopContext("value"); err != nil {
+ return err
+ }
+ } else {
+ stringValue := fmt.Sprintf("%v", elem)
+ if err := writeBuffer.WriteString("value", uint32(len(stringValue)*8), "UTF-8", stringValue); err != nil {
+ return err
+ }
+ }
+ }
+ }
+ if err := writeBuffer.PopContext("subscriptionHandles", utils.WithRenderAsList(true)); err != nil {
return err
}
- if err := writeBuffer.PopContext("PlcUnsubscriptionResponse"); err != nil {
+ if err := writeBuffer.PopContext("PlcUnsubscriptionRequestBuilder"); err != nil {
return err
}
return nil
}
-func (d *DefaultPlcUnsubscriptionResponse) String() string {
+func (d *DefaultPlcUnsubscriptionRequestBuilder) String() string {
writeBuffer := utils.NewWriteBufferBoxBasedWithOptions(true, true)
if err := writeBuffer.WriteSerializable(context.Background(), d); err != nil {
return err.Error()
diff --git a/plc4go/spi/model/DefaultPlcUnsubscriptionRequest_plc4xgen.go b/plc4go/spi/model/DefaultPlcUnsubscriptionRequest_plc4xgen.go
index f11f2de3ca..5c6e624738 100644
--- a/plc4go/spi/model/DefaultPlcUnsubscriptionRequest_plc4xgen.go
+++ b/plc4go/spi/model/DefaultPlcUnsubscriptionRequest_plc4xgen.go
@@ -43,7 +43,50 @@ func (d *DefaultPlcUnsubscriptionRequest) SerializeWithWriteBuffer(ctx context.C
return err
}
- if err := writeBuffer.WriteString("implementMe", uint32(len(d.implementMe)*8), "UTF-8", d.implementMe); err != nil {
+ if d.subscriber != nil {
+ if serializableField, ok := d.subscriber.(utils.Serializable); ok {
+ if err := writeBuffer.PushContext("subscriber"); err != nil {
+ return err
+ }
+ if err := serializableField.SerializeWithWriteBuffer(ctx, writeBuffer); err != nil {
+ return err
+ }
+ if err := writeBuffer.PopContext("subscriber"); err != nil {
+ return err
+ }
+ } else {
+ stringValue := fmt.Sprintf("%v", d.subscriber)
+ if err := writeBuffer.WriteString("subscriber", uint32(len(stringValue)*8), "UTF-8", stringValue); err != nil {
+ return err
+ }
+ }
+ }
+ if err := writeBuffer.PushContext("subscriptionHandles", utils.WithRenderAsList(true)); err != nil {
+ return err
+ }
+ for _, elem := range d.subscriptionHandles {
+ var elem any = elem
+
+ if elem != nil {
+ if serializableField, ok := elem.(utils.Serializable); ok {
+ if err := writeBuffer.PushContext("value"); err != nil {
+ return err
+ }
+ if err := serializableField.SerializeWithWriteBuffer(ctx, writeBuffer); err != nil {
+ return err
+ }
+ if err := writeBuffer.PopContext("value"); err != nil {
+ return err
+ }
+ } else {
+ stringValue := fmt.Sprintf("%v", elem)
+ if err := writeBuffer.WriteString("value", uint32(len(stringValue)*8), "UTF-8", stringValue); err != nil {
+ return err
+ }
+ }
+ }
+ }
+ if err := writeBuffer.PopContext("subscriptionHandles", utils.WithRenderAsList(true)); err != nil {
return err
}
if err := writeBuffer.PopContext("PlcUnsubscriptionRequest"); err != nil {
diff --git a/plc4go/spi/model/DefaultPlcUnsubscriptionResponse.go b/plc4go/spi/model/DefaultPlcUnsubscriptionResponse.go
index 5a54dfe63c..ef153469b9 100644
--- a/plc4go/spi/model/DefaultPlcUnsubscriptionResponse.go
+++ b/plc4go/spi/model/DefaultPlcUnsubscriptionResponse.go
@@ -25,10 +25,14 @@ var _ apiModel.PlcUnsubscriptionResponse = &DefaultPlcUnsubscriptionResponse{}
//go:generate go run ../../tools/plc4xgenerator/gen.go -type=DefaultPlcUnsubscriptionResponse
type DefaultPlcUnsubscriptionResponse struct {
- // TODO: implement
- implementMe string
+ request apiModel.PlcUnsubscriptionRequest `ignore:"true"`
+ none string
}
-func NewDefaultPlcUnsubscriptionResponse() *DefaultPlcUnsubscriptionResponse {
- return &DefaultPlcUnsubscriptionResponse{}
+func (d *DefaultPlcUnsubscriptionResponse) GetRequest() apiModel.PlcUnsubscriptionRequest {
+ return d.request
+}
+
+func NewDefaultPlcUnsubscriptionResponse(request apiModel.PlcUnsubscriptionRequest) *DefaultPlcUnsubscriptionResponse {
+ return &DefaultPlcUnsubscriptionResponse{request: request}
}
diff --git a/plc4go/spi/model/DefaultPlcUnsubscriptionResponse_plc4xgen.go b/plc4go/spi/model/DefaultPlcUnsubscriptionResponse_plc4xgen.go
index ddb8dc6798..c3ee037df8 100644
--- a/plc4go/spi/model/DefaultPlcUnsubscriptionResponse_plc4xgen.go
+++ b/plc4go/spi/model/DefaultPlcUnsubscriptionResponse_plc4xgen.go
@@ -43,7 +43,7 @@ func (d *DefaultPlcUnsubscriptionResponse) SerializeWithWriteBuffer(ctx context.
return err
}
- if err := writeBuffer.WriteString("implementMe", uint32(len(d.implementMe)*8), "UTF-8", d.implementMe); err != nil {
+ if err := writeBuffer.WriteString("none", uint32(len(d.none)*8), "UTF-8", d.none); err != nil {
return err
}
if err := writeBuffer.PopContext("PlcUnsubscriptionResponse"); err != nil {
diff --git a/plc4go/spi/model/render_test.go b/plc4go/spi/model/render_test.go
index 3db88192d9..94081366b5 100644
--- a/plc4go/spi/model/render_test.go
+++ b/plc4go/spi/model/render_test.go
@@ -313,8 +313,8 @@ func TestRenderTestCustom(t *testing.T) {
},
{
sut: NewDefaultPlcUnsubscriptionRequestResult(
- NewDefaultPlcUnsubscriptionRequest(),
- NewDefaultPlcUnsubscriptionResponse(),
+ NewDefaultPlcUnsubscriptionRequest(nil),
+ NewDefaultPlcUnsubscriptionResponse(nil),
nil,
).(interface { // TODO: workaround
fmt.Stringer