plc4go/internal/opcua/SubscriptionHandle.go - plc4x - Git at Google

 /*
  * 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 opcua

 import (
 	"context"
 	"encoding/binary"
 	"slices"
 	"sync"
 	"sync/atomic"
 	"time"

 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"

 	apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
 	readWriteModel "github.com/apache/plc4x/plc4go/protocols/opcua/readwrite/model"
 	spiModel "github.com/apache/plc4x/plc4go/spi/model"
 	"github.com/apache/plc4x/plc4go/spi/utils"
 )

 type SubscriptionHandle struct {
 	*spiModel.DefaultPlcSubscriptionHandle
 	plcSubscriber       *Subscriber
 	connection          *Connection
 	subscriptionRequest apiModel.PlcSubscriptionRequest
 	tagNames            []string
 	consumers           []apiModel.PlcSubscriptionEventConsumer
 	subscriptionId      uint32
 	cycleTime           time.Duration
 	revisedCycleTime    time.Duration
 	clientHandles       atomic.Uint32

 	destroy      atomic.Bool
 	subscriberWg sync.WaitGroup
 	complete     bool

 	wg sync.WaitGroup // use to track spawned go routines

 	log zerolog.Logger
 }

 func NewSubscriptionHandle(log zerolog.Logger, subscriber *Subscriber, connection *Connection, subscriptionRequest apiModel.PlcSubscriptionRequest, subscriptionId uint32, cycleTime time.Duration) *SubscriptionHandle {
 	s := &SubscriptionHandle{
 		plcSubscriber:       subscriber,
 		connection:          connection,
 		subscriptionRequest: subscriptionRequest,
 		tagNames:            subscriptionRequest.GetTagNames(),
 		subscriptionId:      subscriptionId,
 		cycleTime:           cycleTime,
 		log:                 log,
 	}
 	s.clientHandles.Store(1)
 	s.DefaultPlcSubscriptionHandle = spiModel.NewDefaultPlcSubscriptionHandleWithHandleToRegister(subscriber, s)
 	_, err := s.onSubscribeCreateMonitoredItemsRequest()
 	if err != nil {
 		subscriber.onDisconnect()
 	}
 	s.startSubscriber()
 	return s
 }

 func (h *SubscriptionHandle) onSubscribeCreateMonitoredItemsRequest() (readWriteModel.CreateMonitoredItemsResponse, error) {
 	requestList := make([]readWriteModel.MonitoredItemCreateRequest, len(h.tagNames))

 	for _, tagName := range h.tagNames {
 		tagDefaultPlcSubscription := h.subscriptionRequest.GetTag(tagName)

 		idNode, err := generateNodeId(tagDefaultPlcSubscription.(Tag))
 		if err != nil {
 			return nil, errors.Wrap(err, "error genertating node id")
 		}

 		readValueId := readWriteModel.NewReadValueId(
 			idNode,
 			0xD,
 			NULL_STRING,
 			readWriteModel.NewQualifiedName(0, NULL_STRING))

 		var monitoringMode readWriteModel.MonitoringMode
 		switch tagDefaultPlcSubscription.GetPlcSubscriptionType() {
 		case apiModel.SubscriptionCyclic:
 			monitoringMode = readWriteModel.MonitoringMode_monitoringModeSampling
 		case apiModel.SubscriptionChangeOfState:
 			monitoringMode = readWriteModel.MonitoringMode_monitoringModeReporting
 		case apiModel.SubscriptionEvent:
 			monitoringMode = readWriteModel.MonitoringMode_monitoringModeReporting
 		default:
 			monitoringMode = readWriteModel.MonitoringMode_monitoringModeReporting
 		}

 		clientHandle := h.clientHandles.Add(1) - 1

 		parameters := readWriteModel.NewMonitoringParameters(
 			clientHandle,
 			float64(h.cycleTime),  // sampling interval
 			NULL_EXTENSION_OBJECT, // filter, null means use default
 			1,                     // queue size
 			true,                  // discard oldest
 		)

 		request := readWriteModel.NewMonitoredItemCreateRequest(
 			readValueId, monitoringMode, parameters)

 		requestList = append(requestList, request)
 	}

 	requestHeader := readWriteModel.NewRequestHeader(h.connection.channel.getAuthenticationToken(),
 		h.connection.channel.getCurrentDateTime(),
 		h.connection.channel.getRequestHandle(),
 		0,
 		NULL_STRING,
 		REQUEST_TIMEOUT_LONG,
 		NULL_EXTENSION_OBJECT)

 	createMonitoredItemsRequest := readWriteModel.NewCreateMonitoredItemsRequest(
 		requestHeader,
 		h.subscriptionId,
 		readWriteModel.TimestampsToReturn_timestampsToReturnBoth,
 		requestList,
 	)

 	identifier := createMonitoredItemsRequest.GetExtensionId()
 	expandedNodeId := readWriteModel.NewExpandedNodeId(false, //Namespace Uri Specified
 		false, //Server Index Specified
 		readWriteModel.NewNodeIdFourByte(0, uint16(identifier)),
 		nil,
 		nil)

 	extObject := readWriteModel.NewRootExtensionObject(
 		expandedNodeId,
 		createMonitoredItemsRequest,
 	)

 	ctx, cancel := context.WithTimeout(context.Background(), REQUEST_TIMEOUT)
 	defer cancel()
 	buffer := utils.NewWriteBufferByteBased(utils.WithByteOrderForByteBasedBuffer(binary.LittleEndian))
 	if err := extObject.SerializeWithWriteBuffer(ctx, buffer); err != nil {
 		return nil, errors.Wrapf(err, "Unable to serialise the ReadRequest")
 	}

 	responseChan := make(chan readWriteModel.CreateMonitoredItemsResponse, 100) // TODO: bit oversized to not block anything. Discards errors
 	errorChan := make(chan error, 100)                                          // TODO: bit oversized to not block anything. Discards errors
 	consumer := func(opcuaResponse []byte) {
 		unknownExtensionObject, err := readWriteModel.ExtensionObjectParseWithBuffer[readWriteModel.ExtensionObject](ctx, utils.NewReadBufferByteBased(opcuaResponse, utils.WithByteOrderForReadBufferByteBased(binary.LittleEndian)), false)
 		if err != nil {
 			errorChan <- errors.Wrapf(err, "Unable to read the reply")
 			return
 		}
 		var responseMessage readWriteModel.CreateMonitoredItemsResponse
 		switch unknownExtensionObject := unknownExtensionObject.(type) {
 		case readWriteModel.CreateMonitoredItemsResponse:
 			responseMessage = unknownExtensionObject
 		case readWriteModel.ServiceFault:
 			serviceFault := unknownExtensionObject
 			header := serviceFault.GetResponseHeader().(readWriteModel.ResponseHeader)
 			errorChan <- errors.Errorf("Subscription ServiceFault returned from server with error code,  '%s'", header.GetServiceResult())
 			h.plcSubscriber.onDisconnect()
 			return
 		default:
 			errorChan <- errors.Errorf("Unexpected type %T received", unknownExtensionObject)
 			h.plcSubscriber.onDisconnect()
 			return
 		}

 		array := make([]readWriteModel.MonitoredItemCreateResult, len(responseMessage.GetResults()))
 		for i, definition := range responseMessage.GetResults() {
 			array[i] = definition.(readWriteModel.MonitoredItemCreateResult)
 		}
 		for index, arrayLength := 0, len(array); index < arrayLength; index++ {
 			result := array[index]
 			if code, ok := readWriteModel.OpcuaStatusCodeByValue(result.GetStatusCode().GetStatusCode()); !ok || code != readWriteModel.OpcuaStatusCode_Good {
 				h.log.Error().Str("tag", h.tagNames[index]).Msg("Invalid Tag, subscription created without this tag")
 			} else {
 				h.log.Debug().Str("tag", h.tagNames[index]).Msg("Tag was added to the subscription")
 			}
 		}
 		responseChan <- responseMessage
 	}

 	errorDispatcher := func(err error) {
 		errorChan <- errors.Wrap(err, "error received")
 	}

 	h.connection.channel.submit(ctx, h.connection.messageCodec, errorDispatcher, consumer, buffer)

 	select {
 	case response := <-responseChan:
 		return response, nil
 	case err := <-errorChan:
 		return nil, errors.Wrap(err, "error received")
 	case <-ctx.Done():
 		return nil, errors.Wrap(ctx.Err(), "context ended")
 	}
 }

 /**
  * startSubscriber Main subscriber loop. For subscription, we still need to send a request the server on every cycle.
  * Which includes a request for an update of the previously agreed upon list of tags.
  * The server will respond at most once every cycle.
  */
 func (h *SubscriptionHandle) startSubscriber() {
 	h.log.Trace().Msg("Starting Subscription")

 	h.subscriberWg.Add(1)
 	h.wg.Add(1)
 	go func() {
 		defer h.wg.Done()
 		defer h.subscriberWg.Done()

 		var outstandingAcknowledgements []readWriteModel.SubscriptionAcknowledgement
 		var outstandingRequests []uint32
 		for !h.destroy.Load() {

 			requestHandle := h.connection.channel.getRequestHandle()

 			//If we are waiting on a response and haven't received one, just wait until we do. A keep alive will be sent out eventually
 			if len(outstandingRequests) <= 1 {
 				requestHeader := readWriteModel.NewRequestHeader(h.connection.channel.getAuthenticationToken(),
 					h.connection.channel.getCurrentDateTime(),
 					requestHandle,
 					0,
 					NULL_STRING,
 					uint32(h.revisedCycleTime*10),
 					NULL_EXTENSION_OBJECT)

 				//Make a copy of the outstanding requests, so it isn't modified while we are putting the ack list together.
 				acks := slices.Clone(outstandingAcknowledgements)
 				ackLength := len(acks)
 				if ackLength == 0 {
 					ackLength = -1
 				}
 				{ // golang version of remove all
 					tmpOutstandingAcknowledgements := map[readWriteModel.SubscriptionAcknowledgement]bool{}
 					for _, acknowledgement := range outstandingAcknowledgements {
 						tmpOutstandingAcknowledgements[acknowledgement] = true
 					}
 					for _, ack := range acks {
 						delete(tmpOutstandingAcknowledgements, ack)
 					}
 					outstandingAcknowledgements = make([]readWriteModel.SubscriptionAcknowledgement, len(tmpOutstandingAcknowledgements))
 					count := 0
 					for ack := range tmpOutstandingAcknowledgements {
 						outstandingAcknowledgements[count] = ack
 						count++
 					}
 				}

 				publishRequest := readWriteModel.NewPublishRequest(
 					requestHeader,
 					acks,
 				)

 				identifier := publishRequest.GetExtensionId()
 				extExpandedNodeId := readWriteModel.NewExpandedNodeId(false, //Namespace Uri Specified
 					false, //Server Index Specified
 					readWriteModel.NewNodeIdFourByte(0, uint16(identifier)),
 					nil,
 					nil)

 				extObject := readWriteModel.NewRootExtensionObject(
 					extExpandedNodeId,
 					publishRequest,
 				)

 				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")
 					continue
 				}

 				consumer := func(opcuaResponse []byte) {
 					var responseMessage readWriteModel.PublishResponse
 					var serviceFault readWriteModel.ServiceFault
 					unknownExtensionObject, err := readWriteModel.ExtensionObjectParseWithBuffer[readWriteModel.ExtensionObject](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()
 						return
 					}
 					switch unknownExtensionObject := unknownExtensionObject.(type) {
 					case readWriteModel.PublishResponse:
 						responseMessage = unknownExtensionObject
 					case readWriteModel.ServiceFault:
 						serviceFault = unknownExtensionObject
 						header := serviceFault.GetResponseHeader().(readWriteModel.ResponseHeader)
 						h.log.Debug().
 							Stringer("serviceResult", header.GetServiceResult()).
 							Msg("Subscription ServiceFault returned from server with error code, ignoring as it is probably just a result of a Delete Subscription Request")
 						//h.plcSubscriber.onDisconnect(context);
 						return
 					}
 					if serviceFault == nil {
 						handle := responseMessage.GetResponseHeader().(readWriteModel.ResponseHeader).GetRequestHandle()
 						index := 0
 						for i, request := range outstandingRequests {
 							if request == handle {
 								index = i
 							}
 						}
 						outstandingRequests = append(outstandingRequests[:index], outstandingRequests[index+1:]...)

 						for _, availableSequenceNumber := range responseMessage.GetAvailableSequenceNumbers() {
 							outstandingAcknowledgements = append(outstandingAcknowledgements, readWriteModel.NewSubscriptionAcknowledgement(h.subscriptionId, availableSequenceNumber))
 						}

 						for _, notificationMessage := range responseMessage.GetNotificationMessage().(readWriteModel.NotificationMessage).GetNotificationData() {
 							notification := notificationMessage.GetBody()
 							if notification, ok := notification.(readWriteModel.DataChangeNotification); ok {
 								h.log.Trace().Msg("Found a Data Change notification")
 								items := notification.GetMonitoredItems()
 								monitoredNoticiations := make([]readWriteModel.MonitoredItemNotification, len(items))
 								for i, item := range items {
 									monitoredNoticiations[i] = item.(readWriteModel.MonitoredItemNotification)
 								}
 								h.onSubscriptionValue(monitoredNoticiations)
 							} else {
 								h.log.Warn().Msg("Unsupported Notification type")
 							}
 						}
 					}
 				}

 				errorDispatcher := func(err error) {
 					h.log.Error().Err(err).Msg("error received")
 					h.plcSubscriber.onDisconnect()
 				}

 				h.connection.channel.submit(ctx, h.connection.messageCodec, errorDispatcher, consumer, buffer)
 			}
 			//Put the subscriber loop to sleep for the rest of the cycle.
 			time.Sleep(h.revisedCycleTime)
 		}
 		//Wait for any outstanding responses to arrive, using the request timeout length
 		//sleep(this.revisedCycleTime * 10);
 		h.complete = true
 	}()
 }

 // stopSubscriber stops the subscriber either on disconnect or on error
 func (h *SubscriptionHandle) stopSubscriber() {
 	h.destroy.Store(true)

 	requestHandle := h.connection.channel.getRequestHandle()

 	requestHeader := readWriteModel.NewRequestHeader(
 		h.connection.channel.getAuthenticationToken(),
 		h.connection.channel.getCurrentDateTime(),
 		requestHandle,
 		0,
 		NULL_STRING,
 		uint32(h.revisedCycleTime*10),
 		NULL_EXTENSION_OBJECT,
 	)

 	subscriptions := []uint32{h.subscriptionId}
 	deleteSubscriptionrequest := readWriteModel.NewDeleteSubscriptionsRequest(requestHeader,
 		subscriptions,
 	)

 	identifier := deleteSubscriptionrequest.GetExtensionId()
 	extExpandedNodeId := readWriteModel.NewExpandedNodeId(false, //Namespace Uri Specified
 		false, //Server Index Specified
 		readWriteModel.NewNodeIdFourByte(0, uint16(identifier)),
 		nil,
 		nil,
 	)

 	extObject := readWriteModel.NewRootExtensionObject(
 		extExpandedNodeId,
 		deleteSubscriptionrequest,
 	)

 	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[readWriteModel.ExtensionObject](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()
 			return
 		}
 		switch unknownExtensionObject := unknownExtensionObject.(type) {
 		case readWriteModel.DeleteSubscriptionsResponse:
 			responseMessage = unknownExtensionObject
 		case readWriteModel.ServiceFault:
 			serviceFault := unknownExtensionObject
 			header := serviceFault.GetResponseHeader().(readWriteModel.ResponseHeader)
 			h.log.Debug().
 				Stringer("serviceResult", header.GetServiceResult()).
 				Msg("Subscription ServiceFault returned from server with error code, ignoring as it is probably just a result of a Delete Subscription Request")
 			return
 		}
 		h.log.Debug().Stringer("responseMessage", responseMessage).Msg("Received response")
 	}

 	errorDispatcher := func(err error) {
 		h.log.Error().Err(err).Msg("error received")
 		h.plcSubscriber.onDisconnect()
 	}

 	h.connection.channel.submit(ctx, h.connection.messageCodec, errorDispatcher, consumer, buffer)
 }

 /**
  * onSubscriptionValue Receive the returned values from the OPCUA server and format it so that it can be received by the PLC4X client.
  *
  * @param values - array of data values to be sent to the client.
  */
 func (h *SubscriptionHandle) onSubscriptionValue(values []readWriteModel.MonitoredItemNotification) {
 	var tagNameList []string
 	var dataValues []readWriteModel.DataValue
 	for _, value := range values {
 		tagNameList = append(tagNameList, h.tagNames[value.GetClientHandle()-1])
 		dataValues = append(dataValues, value.GetValue())
 	}
 	_, responseCodes, responseValues := readResponse(h.log, nil, tagNameList, dataValues)
 	// TODO: big oof. Where do we get all those nil values from?
 	event := spiModel.NewDefaultPlcSubscriptionEvent(nil, nil, nil, nil, responseCodes, responseValues)

 	for _, consumer := range h.consumers {
 		consumer(event)
 	}
 }
	/*
	* 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 opcua

	import (
	"context"
	"encoding/binary"
	"slices"
	"sync"
	"sync/atomic"
	"time"

	"github.com/pkg/errors"
	"github.com/rs/zerolog"

	apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
	readWriteModel "github.com/apache/plc4x/plc4go/protocols/opcua/readwrite/model"
	spiModel "github.com/apache/plc4x/plc4go/spi/model"
	"github.com/apache/plc4x/plc4go/spi/utils"
	)

	type SubscriptionHandle struct {
	*spiModel.DefaultPlcSubscriptionHandle
	plcSubscriber *Subscriber
	connection *Connection
	subscriptionRequest apiModel.PlcSubscriptionRequest
	tagNames []string
	consumers []apiModel.PlcSubscriptionEventConsumer
	subscriptionId uint32
	cycleTime time.Duration
	revisedCycleTime time.Duration
	clientHandles atomic.Uint32

	destroy atomic.Bool
	subscriberWg sync.WaitGroup
	complete bool

	wg sync.WaitGroup // use to track spawned go routines

	log zerolog.Logger
	}

	func NewSubscriptionHandle(log zerolog.Logger, subscriber Subscriber, connection Connection, subscriptionRequest apiModel.PlcSubscriptionRequest, subscriptionId uint32, cycleTime time.Duration) *SubscriptionHandle {
	s := &SubscriptionHandle{
	plcSubscriber: subscriber,
	connection: connection,
	subscriptionRequest: subscriptionRequest,
	tagNames: subscriptionRequest.GetTagNames(),
	subscriptionId: subscriptionId,
	cycleTime: cycleTime,
	log: log,
	}
	s.clientHandles.Store(1)
	s.DefaultPlcSubscriptionHandle = spiModel.NewDefaultPlcSubscriptionHandleWithHandleToRegister(subscriber, s)
	_, err := s.onSubscribeCreateMonitoredItemsRequest()
	if err != nil {
	subscriber.onDisconnect()
	}
	s.startSubscriber()
	return s
	}

	func (h *SubscriptionHandle) onSubscribeCreateMonitoredItemsRequest() (readWriteModel.CreateMonitoredItemsResponse, error) {
	requestList := make([]readWriteModel.MonitoredItemCreateRequest, len(h.tagNames))

	for _, tagName := range h.tagNames {
	tagDefaultPlcSubscription := h.subscriptionRequest.GetTag(tagName)

	idNode, err := generateNodeId(tagDefaultPlcSubscription.(Tag))
	if err != nil {
	return nil, errors.Wrap(err, "error genertating node id")
	}

	readValueId := readWriteModel.NewReadValueId(
	idNode,
	0xD,
	NULL_STRING,
	readWriteModel.NewQualifiedName(0, NULL_STRING))

	var monitoringMode readWriteModel.MonitoringMode
	switch tagDefaultPlcSubscription.GetPlcSubscriptionType() {
	case apiModel.SubscriptionCyclic:
	monitoringMode = readWriteModel.MonitoringMode_monitoringModeSampling
	case apiModel.SubscriptionChangeOfState:
	monitoringMode = readWriteModel.MonitoringMode_monitoringModeReporting
	case apiModel.SubscriptionEvent:
	monitoringMode = readWriteModel.MonitoringMode_monitoringModeReporting
	default:
	monitoringMode = readWriteModel.MonitoringMode_monitoringModeReporting
	}

	clientHandle := h.clientHandles.Add(1) - 1

	parameters := readWriteModel.NewMonitoringParameters(
	clientHandle,
	float64(h.cycleTime), // sampling interval
	NULL_EXTENSION_OBJECT, // filter, null means use default
	1, // queue size
	true, // discard oldest
	)

	request := readWriteModel.NewMonitoredItemCreateRequest(
	readValueId, monitoringMode, parameters)

	requestList = append(requestList, request)
	}

	requestHeader := readWriteModel.NewRequestHeader(h.connection.channel.getAuthenticationToken(),
	h.connection.channel.getCurrentDateTime(),
	h.connection.channel.getRequestHandle(),
	0,
	NULL_STRING,
	REQUEST_TIMEOUT_LONG,
	NULL_EXTENSION_OBJECT)

	createMonitoredItemsRequest := readWriteModel.NewCreateMonitoredItemsRequest(
	requestHeader,
	h.subscriptionId,
	readWriteModel.TimestampsToReturn_timestampsToReturnBoth,
	requestList,
	)

	identifier := createMonitoredItemsRequest.GetExtensionId()
	expandedNodeId := readWriteModel.NewExpandedNodeId(false, //Namespace Uri Specified
	false, //Server Index Specified
	readWriteModel.NewNodeIdFourByte(0, uint16(identifier)),
	nil,
	nil)

	extObject := readWriteModel.NewRootExtensionObject(
	expandedNodeId,
	createMonitoredItemsRequest,
	)

	ctx, cancel := context.WithTimeout(context.Background(), REQUEST_TIMEOUT)
	defer cancel()
	buffer := utils.NewWriteBufferByteBased(utils.WithByteOrderForByteBasedBuffer(binary.LittleEndian))
	if err := extObject.SerializeWithWriteBuffer(ctx, buffer); err != nil {
	return nil, errors.Wrapf(err, "Unable to serialise the ReadRequest")
	}

	responseChan := make(chan readWriteModel.CreateMonitoredItemsResponse, 100) // TODO: bit oversized to not block anything. Discards errors
	errorChan := make(chan error, 100) // TODO: bit oversized to not block anything. Discards errors
	consumer := func(opcuaResponse []byte) {
	unknownExtensionObject, err := readWriteModel.ExtensionObjectParseWithBuffer[readWriteModel.ExtensionObject](ctx, utils.NewReadBufferByteBased(opcuaResponse, utils.WithByteOrderForReadBufferByteBased(binary.LittleEndian)), false)
	if err != nil {
	errorChan <- errors.Wrapf(err, "Unable to read the reply")
	return
	}
	var responseMessage readWriteModel.CreateMonitoredItemsResponse
	switch unknownExtensionObject := unknownExtensionObject.(type) {
	case readWriteModel.CreateMonitoredItemsResponse:
	responseMessage = unknownExtensionObject
	case readWriteModel.ServiceFault:
	serviceFault := unknownExtensionObject
	header := serviceFault.GetResponseHeader().(readWriteModel.ResponseHeader)
	errorChan <- errors.Errorf("Subscription ServiceFault returned from server with error code, '%s'", header.GetServiceResult())
	h.plcSubscriber.onDisconnect()
	return
	default:
	errorChan <- errors.Errorf("Unexpected type %T received", unknownExtensionObject)
	h.plcSubscriber.onDisconnect()
	return
	}

	array := make([]readWriteModel.MonitoredItemCreateResult, len(responseMessage.GetResults()))
	for i, definition := range responseMessage.GetResults() {
	array[i] = definition.(readWriteModel.MonitoredItemCreateResult)
	}
	for index, arrayLength := 0, len(array); index < arrayLength; index++ {
	result := array[index]
	if code, ok := readWriteModel.OpcuaStatusCodeByValue(result.GetStatusCode().GetStatusCode()); !ok \|\| code != readWriteModel.OpcuaStatusCode_Good {
	h.log.Error().Str("tag", h.tagNames[index]).Msg("Invalid Tag, subscription created without this tag")
	} else {
	h.log.Debug().Str("tag", h.tagNames[index]).Msg("Tag was added to the subscription")
	}
	}
	responseChan <- responseMessage
	}

	errorDispatcher := func(err error) {
	errorChan <- errors.Wrap(err, "error received")
	}

	h.connection.channel.submit(ctx, h.connection.messageCodec, errorDispatcher, consumer, buffer)

	select {
	case response := <-responseChan:
	return response, nil
	case err := <-errorChan:
	return nil, errors.Wrap(err, "error received")
	case <-ctx.Done():
	return nil, errors.Wrap(ctx.Err(), "context ended")
	}
	}

	/**
	* startSubscriber Main subscriber loop. For subscription, we still need to send a request the server on every cycle.
	* Which includes a request for an update of the previously agreed upon list of tags.
	* The server will respond at most once every cycle.
	*/
	func (h *SubscriptionHandle) startSubscriber() {
	h.log.Trace().Msg("Starting Subscription")

	h.subscriberWg.Add(1)
	h.wg.Add(1)
	go func() {
	defer h.wg.Done()
	defer h.subscriberWg.Done()

	var outstandingAcknowledgements []readWriteModel.SubscriptionAcknowledgement
	var outstandingRequests []uint32
	for !h.destroy.Load() {

	requestHandle := h.connection.channel.getRequestHandle()

	//If we are waiting on a response and haven't received one, just wait until we do. A keep alive will be sent out eventually
	if len(outstandingRequests) <= 1 {
	requestHeader := readWriteModel.NewRequestHeader(h.connection.channel.getAuthenticationToken(),
	h.connection.channel.getCurrentDateTime(),
	requestHandle,
	0,
	NULL_STRING,
	uint32(h.revisedCycleTime*10),
	NULL_EXTENSION_OBJECT)

	//Make a copy of the outstanding requests, so it isn't modified while we are putting the ack list together.
	acks := slices.Clone(outstandingAcknowledgements)
	ackLength := len(acks)
	if ackLength == 0 {
	ackLength = -1
	}
	{ // golang version of remove all
	tmpOutstandingAcknowledgements := map[readWriteModel.SubscriptionAcknowledgement]bool{}
	for _, acknowledgement := range outstandingAcknowledgements {
	tmpOutstandingAcknowledgements[acknowledgement] = true
	}
	for _, ack := range acks {
	delete(tmpOutstandingAcknowledgements, ack)
	}
	outstandingAcknowledgements = make([]readWriteModel.SubscriptionAcknowledgement, len(tmpOutstandingAcknowledgements))
	count := 0
	for ack := range tmpOutstandingAcknowledgements {
	outstandingAcknowledgements[count] = ack
	count++
	}
	}

	publishRequest := readWriteModel.NewPublishRequest(
	requestHeader,
	acks,
	)

	identifier := publishRequest.GetExtensionId()
	extExpandedNodeId := readWriteModel.NewExpandedNodeId(false, //Namespace Uri Specified
	false, //Server Index Specified
	readWriteModel.NewNodeIdFourByte(0, uint16(identifier)),
	nil,
	nil)

	extObject := readWriteModel.NewRootExtensionObject(
	extExpandedNodeId,
	publishRequest,
	)

	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")
	continue
	}

	consumer := func(opcuaResponse []byte) {
	var responseMessage readWriteModel.PublishResponse
	var serviceFault readWriteModel.ServiceFault
	unknownExtensionObject, err := readWriteModel.ExtensionObjectParseWithBuffer[readWriteModel.ExtensionObject](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()
	return
	}
	switch unknownExtensionObject := unknownExtensionObject.(type) {
	case readWriteModel.PublishResponse:
	responseMessage = unknownExtensionObject
	case readWriteModel.ServiceFault:
	serviceFault = unknownExtensionObject
	header := serviceFault.GetResponseHeader().(readWriteModel.ResponseHeader)
	h.log.Debug().
	Stringer("serviceResult", header.GetServiceResult()).
	Msg("Subscription ServiceFault returned from server with error code, ignoring as it is probably just a result of a Delete Subscription Request")
	//h.plcSubscriber.onDisconnect(context);
	return
	}
	if serviceFault == nil {
	handle := responseMessage.GetResponseHeader().(readWriteModel.ResponseHeader).GetRequestHandle()
	index := 0
	for i, request := range outstandingRequests {
	if request == handle {
	index = i
	}
	}
	outstandingRequests = append(outstandingRequests[:index], outstandingRequests[index+1:]...)

	for _, availableSequenceNumber := range responseMessage.GetAvailableSequenceNumbers() {
	outstandingAcknowledgements = append(outstandingAcknowledgements, readWriteModel.NewSubscriptionAcknowledgement(h.subscriptionId, availableSequenceNumber))
	}

	for _, notificationMessage := range responseMessage.GetNotificationMessage().(readWriteModel.NotificationMessage).GetNotificationData() {
	notification := notificationMessage.GetBody()
	if notification, ok := notification.(readWriteModel.DataChangeNotification); ok {
	h.log.Trace().Msg("Found a Data Change notification")
	items := notification.GetMonitoredItems()
	monitoredNoticiations := make([]readWriteModel.MonitoredItemNotification, len(items))
	for i, item := range items {
	monitoredNoticiations[i] = item.(readWriteModel.MonitoredItemNotification)
	}
	h.onSubscriptionValue(monitoredNoticiations)
	} else {
	h.log.Warn().Msg("Unsupported Notification type")
	}
	}
	}
	}

	errorDispatcher := func(err error) {
	h.log.Error().Err(err).Msg("error received")
	h.plcSubscriber.onDisconnect()
	}

	h.connection.channel.submit(ctx, h.connection.messageCodec, errorDispatcher, consumer, buffer)
	}
	//Put the subscriber loop to sleep for the rest of the cycle.
	time.Sleep(h.revisedCycleTime)
	}
	//Wait for any outstanding responses to arrive, using the request timeout length
	//sleep(this.revisedCycleTime * 10);
	h.complete = true
	}()
	}

	// stopSubscriber stops the subscriber either on disconnect or on error
	func (h *SubscriptionHandle) stopSubscriber() {
	h.destroy.Store(true)

	requestHandle := h.connection.channel.getRequestHandle()

	requestHeader := readWriteModel.NewRequestHeader(
	h.connection.channel.getAuthenticationToken(),
	h.connection.channel.getCurrentDateTime(),
	requestHandle,
	0,
	NULL_STRING,
	uint32(h.revisedCycleTime*10),
	NULL_EXTENSION_OBJECT,
	)

	subscriptions := []uint32{h.subscriptionId}
	deleteSubscriptionrequest := readWriteModel.NewDeleteSubscriptionsRequest(requestHeader,
	subscriptions,
	)

	identifier := deleteSubscriptionrequest.GetExtensionId()
	extExpandedNodeId := readWriteModel.NewExpandedNodeId(false, //Namespace Uri Specified
	false, //Server Index Specified
	readWriteModel.NewNodeIdFourByte(0, uint16(identifier)),
	nil,
	nil,
	)

	extObject := readWriteModel.NewRootExtensionObject(
	extExpandedNodeId,
	deleteSubscriptionrequest,
	)

	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[readWriteModel.ExtensionObject](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()
	return
	}
	switch unknownExtensionObject := unknownExtensionObject.(type) {
	case readWriteModel.DeleteSubscriptionsResponse:
	responseMessage = unknownExtensionObject
	case readWriteModel.ServiceFault:
	serviceFault := unknownExtensionObject
	header := serviceFault.GetResponseHeader().(readWriteModel.ResponseHeader)
	h.log.Debug().
	Stringer("serviceResult", header.GetServiceResult()).
	Msg("Subscription ServiceFault returned from server with error code, ignoring as it is probably just a result of a Delete Subscription Request")
	return
	}
	h.log.Debug().Stringer("responseMessage", responseMessage).Msg("Received response")
	}

	errorDispatcher := func(err error) {
	h.log.Error().Err(err).Msg("error received")
	h.plcSubscriber.onDisconnect()
	}

	h.connection.channel.submit(ctx, h.connection.messageCodec, errorDispatcher, consumer, buffer)
	}

	/**
	* onSubscriptionValue Receive the returned values from the OPCUA server and format it so that it can be received by the PLC4X client.
	*
	* @param values - array of data values to be sent to the client.
	*/
	func (h *SubscriptionHandle) onSubscriptionValue(values []readWriteModel.MonitoredItemNotification) {
	var tagNameList []string
	var dataValues []readWriteModel.DataValue
	for _, value := range values {
	tagNameList = append(tagNameList, h.tagNames[value.GetClientHandle()-1])
	dataValues = append(dataValues, value.GetValue())
	}
	_, responseCodes, responseValues := readResponse(h.log, nil, tagNameList, dataValues)
	// TODO: big oof. Where do we get all those nil values from?
	event := spiModel.NewDefaultPlcSubscriptionEvent(nil, nil, nil, nil, responseCodes, responseValues)

	for _, consumer := range h.consumers {
	consumer(event)
	}
	}