refactor(plc4go): use buffered channels when possible

plc4go/internal/ads/Browser.go

@@ -41,7 +41,7 @@ func (m *Connection) Browse(ctx context.Context, browseRequest apiModel.PlcBrows
 }
 
 func (m *Connection) BrowseWithInterceptor(ctx context.Context, browseRequest apiModel.PlcBrowseRequest, interceptor func(result apiModel.PlcBrowseItem) bool) <-chan apiModel.PlcBrowseRequestResult {
-	result := make(chan apiModel.PlcBrowseRequestResult)
+	result := make(chan apiModel.PlcBrowseRequestResult, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {

plc4go/internal/ads/Connection.go

@@ -97,7 +97,7 @@ func (m *Connection) GetConnection() plc4go.PlcConnection {
 
 func (m *Connection) ConnectWithContext(ctx context.Context) <-chan plc4go.PlcConnectionConnectResult {
 	log.Trace().Msg("Connecting")
-	ch := make(chan plc4go.PlcConnectionConnectResult)
+	ch := make(chan plc4go.PlcConnectionConnectResult, 1)
 
 	// Reset the driver context (Actually this should not be required, but just to be on the safe side)
 	m.driverContext.clear()

plc4go/internal/ads/Driver.go

@@ -61,7 +61,7 @@ func (m *Driver) GetConnectionWithContext(ctx context.Context, transportUrl url.
 	transportInstance, err := transport.CreateTransportInstance(transportUrl, options)
 	if err != nil {
 		log.Error().Stringer("transportUrl", &transportUrl).Msgf("We couldn't create a transport instance for port %#v", options["defaultTcpPort"])
-		ch := make(chan plc4go.PlcConnectionConnectResult)
+		ch := make(chan plc4go.PlcConnectionConnectResult, 1)
 		ch <- _default.NewDefaultPlcConnectionConnectResult(nil, errors.New("couldn't initialize transport configuration for given transport url "+transportUrl.String()))
 		return ch
 	}
@@ -73,7 +73,7 @@ func (m *Driver) GetConnectionWithContext(ctx context.Context, transportUrl url.
 	configuration, err := model.ParseFromOptions(options)
 	if err != nil {
 		log.Error().Err(err).Msgf("Invalid options")
-		ch := make(chan plc4go.PlcConnectionConnectResult)
+		ch := make(chan plc4go.PlcConnectionConnectResult, 1)
 		ch <- _default.NewDefaultPlcConnectionConnectResult(nil, errors.Wrap(err, "invalid configuration"))
 		return ch
 	}

plc4go/internal/ads/Interactions.go

@@ -30,7 +30,7 @@ import (
 )
 
 func (m *Connection) ExecuteAdsReadDeviceInfoRequest(ctx context.Context) (model.AdsReadDeviceInfoResponse, error) {
-	responseChannel := make(chan model.AdsReadDeviceInfoResponse)
+	responseChannel := make(chan model.AdsReadDeviceInfoResponse, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {
@@ -71,7 +71,7 @@ func (m *Connection) ExecuteAdsReadDeviceInfoRequest(ctx context.Context) (model
 }
 
 func (m *Connection) ExecuteAdsReadRequest(ctx context.Context, indexGroup uint32, indexOffset uint32, length uint32) (model.AdsReadResponse, error) {
-	responseChannel := make(chan model.AdsReadResponse)
+	responseChannel := make(chan model.AdsReadResponse, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {
@@ -112,7 +112,7 @@ func (m *Connection) ExecuteAdsReadRequest(ctx context.Context, indexGroup uint3
 }
 
 func (m *Connection) ExecuteAdsWriteRequest(ctx context.Context, indexGroup uint32, indexOffset uint32, data []byte) (model.AdsWriteResponse, error) {
-	responseChannel := make(chan model.AdsWriteResponse)
+	responseChannel := make(chan model.AdsWriteResponse, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {
@@ -153,7 +153,7 @@ func (m *Connection) ExecuteAdsWriteRequest(ctx context.Context, indexGroup uint
 }
 
 func (m *Connection) ExecuteAdsReadWriteRequest(ctx context.Context, indexGroup uint32, indexOffset uint32, readLength uint32, items []model.AdsMultiRequestItem, writeData []byte) (model.AdsReadWriteResponse, error) {
-	responseChannel := make(chan model.AdsReadWriteResponse)
+	responseChannel := make(chan model.AdsReadWriteResponse, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {
@@ -194,7 +194,7 @@ func (m *Connection) ExecuteAdsReadWriteRequest(ctx context.Context, indexGroup
 }
 
 func (m *Connection) ExecuteAdsAddDeviceNotificationRequest(ctx context.Context, indexGroup uint32, indexOffset uint32, length uint32, transmissionMode model.AdsTransMode, maxDelay uint32, cycleTime uint32) (model.AdsAddDeviceNotificationResponse, error) {
-	responseChannel := make(chan model.AdsAddDeviceNotificationResponse)
+	responseChannel := make(chan model.AdsAddDeviceNotificationResponse, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {
@@ -235,7 +235,7 @@ func (m *Connection) ExecuteAdsAddDeviceNotificationRequest(ctx context.Context,
 }
 
 func (m *Connection) ExecuteAdsDeleteDeviceNotificationRequest(ctx context.Context, notificationHandle uint32) (model.AdsDeleteDeviceNotificationResponse, error) {
-	responseChannel := make(chan model.AdsDeleteDeviceNotificationResponse)
+	responseChannel := make(chan model.AdsDeleteDeviceNotificationResponse, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {

plc4go/internal/ads/Reader.go

@@ -42,7 +42,7 @@ func (m *Connection) ReadRequestBuilder() apiModel.PlcReadRequestBuilder {
 
 func (m *Connection) Read(ctx context.Context, readRequest apiModel.PlcReadRequest) <-chan apiModel.PlcReadRequestResult {
 	log.Trace().Msg("Reading")
-	result := make(chan apiModel.PlcReadRequestResult)
+	result := make(chan apiModel.PlcReadRequestResult, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {

plc4go/internal/ads/Subscriber.go

@@ -97,7 +97,7 @@ func (m *Connection) Subscribe(ctx context.Context, subscriptionRequest apiModel
 	}
 
 	// Create a new result-channel, which completes as soon as all sub-result-channels have returned
-	globalResultChannel := make(chan apiModel.PlcSubscriptionRequestResult)
+	globalResultChannel := make(chan apiModel.PlcSubscriptionRequestResult, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {
@@ -126,8 +126,8 @@ func (m *Connection) Subscribe(ctx context.Context, subscriptionRequest apiModel
 }
 
 func (m *Connection) subscribe(ctx context.Context, subscriptionRequest apiModel.PlcSubscriptionRequest) <-chan apiModel.PlcSubscriptionRequestResult {
-	responseChan := make(chan apiModel.PlcSubscriptionRequestResult)
-	go func(respChan chan apiModel.PlcSubscriptionRequestResult) {
+	responseChan := make(chan apiModel.PlcSubscriptionRequestResult, 1)
+	go func() {
 		defer func() {
 			if err := recover(); err != nil {
 				responseChan <- internalModel.NewDefaultPlcSubscriptionRequestResult(subscriptionRequest, nil, errors.Errorf("panic-ed %v", err))

plc4go/internal/ads/Writer.go

@@ -41,7 +41,7 @@ func (m *Connection) WriteRequestBuilder() apiModel.PlcWriteRequestBuilder {
 
 func (m *Connection) Write(ctx context.Context, writeRequest apiModel.PlcWriteRequest) <-chan apiModel.PlcWriteRequestResult {
 	log.Trace().Msg("Writing")
-	result := make(chan apiModel.PlcWriteRequestResult)
+	result := make(chan apiModel.PlcWriteRequestResult, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {

plc4go/internal/bacnetip/Connection.go

@@ -78,7 +78,7 @@ func (c *Connection) GetTracer() *spi.Tracer {
 
 func (c *Connection) ConnectWithContext(ctx context.Context) <-chan plc4go.PlcConnectionConnectResult {
 	log.Trace().Msg("Connecting")
-	ch := make(chan plc4go.PlcConnectionConnectResult)
+	ch := make(chan plc4go.PlcConnectionConnectResult, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {

plc4go/internal/bacnetip/Discoverer.go

@@ -84,7 +84,7 @@ func (d *Discoverer) Discover(ctx context.Context, callback func(event apiModel.
 }
 
 func broadcastAndDiscover(ctx context.Context, communicationChannels []communicationChannel, specificOptions *protocolSpecificOptions) (chan receivedBvlcMessage, error) {
-	incomingBVLCChannel := make(chan receivedBvlcMessage, 0)
+	incomingBVLCChannel := make(chan receivedBvlcMessage)
 	for _, communicationChannelInstance := range communicationChannels {
 		// Prepare the discovery packet data
 		{
@@ -192,7 +192,7 @@ func broadcastAndDiscover(ctx context.Context, communicationChannels []communica
 
 		go func(communicationChannelInstance communicationChannel) {
 			for {
-				blockingReadChan := make(chan bool, 0)
+				blockingReadChan := make(chan bool)
 				go func() {
 					buf := make([]byte, 4096)
 					n, addr, err := communicationChannelInstance.broadcastConnection.ReadFrom(buf)

plc4go/internal/bacnetip/Driver.go

@@ -67,10 +67,8 @@ func (m *Driver) GetConnectionWithContext(ctx context.Context, transportUrl url.
 	transport, ok := transports[transportUrl.Scheme]
 	if !ok {
 		log.Error().Stringer("transportUrl", &transportUrl).Msgf("We couldn't find a transport for scheme %s", transportUrl.Scheme)
-		ch := make(chan plc4go.PlcConnectionConnectResult)
-		go func() {
-			ch <- _default.NewDefaultPlcConnectionConnectResult(nil, errors.Errorf("couldn't find transport for given transport url %#v", transportUrl))
-		}()
+		ch := make(chan plc4go.PlcConnectionConnectResult, 1)
+		ch <- _default.NewDefaultPlcConnectionConnectResult(nil, errors.Errorf("couldn't find transport for given transport url %#v", transportUrl))
 		return ch
 	}
 	// Provide a default-port to the transport, which is used, if the user doesn't provide on in the connection string.
@@ -85,19 +83,15 @@ func (m *Driver) GetConnectionWithContext(ctx context.Context, transportUrl url.
 		udpTransport = transport
 	default:
 		log.Error().Stringer("transportUrl", &transportUrl).Msg("Only udp supported at the moment")
-		ch := make(chan plc4go.PlcConnectionConnectResult)
-		go func() {
-			ch <- _default.NewDefaultPlcConnectionConnectResult(nil, errors.Errorf("couldn't find transport for given transport url %#v", transportUrl))
-		}()
+		ch := make(chan plc4go.PlcConnectionConnectResult, 1)
+		ch <- _default.NewDefaultPlcConnectionConnectResult(nil, errors.Errorf("couldn't find transport for given transport url %#v", transportUrl))
 		return ch
 	}
 
 	codec, err := m.applicationManager.getApplicationLayerMessageCodec(udpTransport, transportUrl, options)
 	if err != nil {
-		ch := make(chan plc4go.PlcConnectionConnectResult)
-		go func() {
-			ch <- _default.NewDefaultPlcConnectionConnectResult(nil, errors.Wrap(err, "error getting application layer message codec"))
-		}()
+		ch := make(chan plc4go.PlcConnectionConnectResult, 1)
+		ch <- _default.NewDefaultPlcConnectionConnectResult(nil, errors.Wrap(err, "error getting application layer message codec"))
 		return ch
 	}
 	log.Debug().Msgf("working with codec %#v", codec)

plc4go/internal/bacnetip/Reader.go

@@ -57,7 +57,7 @@ func NewReader(invokeIdGenerator *InvokeIdGenerator, messageCodec spi.MessageCod
 func (m *Reader) Read(ctx context.Context, readRequest apiModel.PlcReadRequest) <-chan apiModel.PlcReadRequestResult {
 	// TODO: handle ctx
 	log.Trace().Msg("Reading")
-	result := make(chan apiModel.PlcReadRequestResult)
+	result := make(chan apiModel.PlcReadRequestResult, 1)
 	go func() {
 		if len(readRequest.GetTagNames()) == 0 {
 			result <- &spiModel.DefaultPlcReadRequestResult{

plc4go/internal/bacnetip/Subscriber.go

@@ -40,8 +40,7 @@ func NewSubscriber(connection *Connection) *Subscriber {
 }
 
 func (m *Subscriber) Subscribe(ctx context.Context, subscriptionRequest apiModel.PlcSubscriptionRequest) <-chan apiModel.PlcSubscriptionRequestResult {
-	// TODO: handle ctx
-	result := make(chan apiModel.PlcSubscriptionRequestResult)
+	result := make(chan apiModel.PlcSubscriptionRequestResult, 1)
 	go func() {
 		internalPlcSubscriptionRequest := subscriptionRequest.(*spiModel.DefaultPlcSubscriptionRequest)
 
@@ -67,7 +66,8 @@ func (m *Subscriber) Subscribe(ctx context.Context, subscriptionRequest apiModel
 
 func (m *Subscriber) Unsubscribe(ctx context.Context, unsubscriptionRequest apiModel.PlcUnsubscriptionRequest) <-chan apiModel.PlcUnsubscriptionRequestResult {
 	// TODO: handle ctx
-	result := make(chan apiModel.PlcUnsubscriptionRequestResult)
+	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 an internal handling of which values to pass where.

plc4go/internal/cbus/Reader.go

@@ -49,7 +49,7 @@ func NewReader(tpduGenerator *AlphaGenerator, messageCodec *MessageCodec, tm spi
 
 func (m *Reader) Read(ctx context.Context, readRequest apiModel.PlcReadRequest) <-chan apiModel.PlcReadRequestResult {
 	log.Trace().Msg("Reading")
-	result := make(chan apiModel.PlcReadRequestResult)
+	result := make(chan apiModel.PlcReadRequestResult, 1)
 	go m.readSync(ctx, readRequest, result)
 	return result
 }

plc4go/internal/cbus/Subscriber.go

@@ -47,7 +47,7 @@ func NewSubscriber(connection *Connection) *Subscriber {
 }
 
 func (m *Subscriber) Subscribe(_ context.Context, subscriptionRequest apiModel.PlcSubscriptionRequest) <-chan apiModel.PlcSubscriptionRequestResult {
-	result := make(chan apiModel.PlcSubscriptionRequestResult)
+	result := make(chan apiModel.PlcSubscriptionRequestResult, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {
@@ -78,7 +78,8 @@ func (m *Subscriber) Subscribe(_ context.Context, subscriptionRequest apiModel.P
 
 func (m *Subscriber) Unsubscribe(ctx context.Context, unsubscriptionRequest apiModel.PlcUnsubscriptionRequest) <-chan apiModel.PlcUnsubscriptionRequestResult {
 	// TODO: handle context
-	result := make(chan apiModel.PlcUnsubscriptionRequestResult)
+	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.

plc4go/internal/cbus/Writer.go

@@ -48,7 +48,7 @@ func NewWriter(tpduGenerator *AlphaGenerator, messageCodec *MessageCodec, tm spi
 
 func (m *Writer) Write(ctx context.Context, writeRequest apiModel.PlcWriteRequest) <-chan apiModel.PlcWriteRequestResult {
 	log.Trace().Msg("Writing")
-	result := make(chan apiModel.PlcWriteRequestResult)
+	result := make(chan apiModel.PlcWriteRequestResult, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {

plc4go/internal/eip/Connection.go

@@ -104,7 +104,7 @@ func (m *Connection) GetMessageCodec() spi.MessageCodec {
 
 func (m *Connection) ConnectWithContext(ctx context.Context) <-chan plc4go.PlcConnectionConnectResult {
 	log.Trace().Msg("Connecting")
-	ch := make(chan plc4go.PlcConnectionConnectResult)
+	ch := make(chan plc4go.PlcConnectionConnectResult, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {
@@ -135,7 +135,7 @@ func (m *Connection) ConnectWithContext(ctx context.Context) <-chan plc4go.PlcCo
 func (m *Connection) Close() <-chan plc4go.PlcConnectionCloseResult {
 	// TODO: use proper context
 	ctx := context.TODO()
-	result := make(chan plc4go.PlcConnectionCloseResult)
+	result := make(chan plc4go.PlcConnectionCloseResult, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {

plc4go/internal/eip/Reader.go

@@ -58,7 +58,7 @@ func NewReader(messageCodec spi.MessageCodec, tm spi.RequestTransactionManager,
 func (m *Reader) Read(ctx context.Context, readRequest apiModel.PlcReadRequest) <-chan apiModel.PlcReadRequestResult {
 	// TODO: handle ctx
 	log.Trace().Msg("Reading")
-	result := make(chan apiModel.PlcReadRequestResult)
+	result := make(chan apiModel.PlcReadRequestResult, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {

plc4go/internal/eip/Writer.go

@@ -55,7 +55,7 @@ func NewWriter(messageCodec spi.MessageCodec, tm spi.RequestTransactionManager,
 
 func (m Writer) Write(ctx context.Context, writeRequest model.PlcWriteRequest) <-chan model.PlcWriteRequestResult {
 	// TODO: handle context
-	result := make(chan model.PlcWriteRequestResult)
+	result := make(chan model.PlcWriteRequestResult, 1)
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {

plc4go/internal/knxnetip/Connection.go

@@ -221,7 +221,7 @@ func (m *Connection) Connect() <-chan plc4go.PlcConnectionConnectResult {
 }
 
 func (m *Connection) ConnectWithContext(ctx context.Context) <-chan plc4go.PlcConnectionConnectResult {
-	result := make(chan plc4go.PlcConnectionConnectResult)
+	result := make(chan plc4go.PlcConnectionConnectResult, 1)
 	sendResult := func(connection plc4go.PlcConnection, err error) {
 		result <- _default.NewDefaultPlcConnectionConnectResult(connection, err)
 	}
@@ -396,7 +396,7 @@ func (m *Connection) BlockingClose() {
 func (m *Connection) Close() <-chan plc4go.PlcConnectionCloseResult {
 	// TODO: use proper context
 	ctx := context.TODO()
-	result := make(chan plc4go.PlcConnectionCloseResult)
+	result := make(chan plc4go.PlcConnectionCloseResult, 1)
 
 	go func() {
 		defer func() {
@@ -459,7 +459,7 @@ func (m *Connection) IsConnected() bool {
 func (m *Connection) Ping() <-chan plc4go.PlcConnectionPingResult {
 	// TODO: use proper context
 	ctx := context.TODO()
-	result := make(chan plc4go.PlcConnectionPingResult)
+	result := make(chan plc4go.PlcConnectionPingResult, 1)
 
 	go func() {
 		defer func() {

plc4go/internal/knxnetip/ConnectionDriverSpecificOperations.go

@@ -45,7 +45,7 @@ import (
 ///////////////////////////////////////////////////////////////////////////////////////////////////////
 
 func (m *Connection) ReadGroupAddress(ctx context.Context, groupAddress []byte, datapointType *driverModel.KnxDatapointType) <-chan KnxReadResult {
-	result := make(chan KnxReadResult)
+	result := make(chan KnxReadResult, 1)
 
 	sendResponse := func(value values.PlcValue, numItems uint8, err error) {
 		timeout := time.NewTimer(time.Millisecond * 10)
@@ -106,7 +106,7 @@ func (m *Connection) ReadGroupAddress(ctx context.Context, groupAddress []byte,
 }
 
 func (m *Connection) DeviceConnect(ctx context.Context, targetAddress driverModel.KnxAddress) <-chan KnxDeviceConnectResult {
-	result := make(chan KnxDeviceConnectResult)
+	result := make(chan KnxDeviceConnectResult, 1)
 
 	sendResponse := func(connection *KnxDeviceConnection, err error) {
 		timeout := time.NewTimer(time.Millisecond * 10)
@@ -201,7 +201,7 @@ func (m *Connection) DeviceConnect(ctx context.Context, targetAddress driverMode
 }
 
 func (m *Connection) DeviceDisconnect(ctx context.Context, targetAddress driverModel.KnxAddress) <-chan KnxDeviceDisconnectResult {
-	result := make(chan KnxDeviceDisconnectResult)
+	result := make(chan KnxDeviceDisconnectResult, 1)
 
 	sendResponse := func(connection *KnxDeviceConnection, err error) {
 		timeout := time.NewTimer(time.Millisecond * 10)
@@ -240,7 +240,7 @@ func (m *Connection) DeviceDisconnect(ctx context.Context, targetAddress driverM
 }
 
 func (m *Connection) DeviceAuthenticate(ctx context.Context, targetAddress driverModel.KnxAddress, buildingKey []byte) <-chan KnxDeviceAuthenticateResult {
-	result := make(chan KnxDeviceAuthenticateResult)
+	result := make(chan KnxDeviceAuthenticateResult, 1)
 
 	sendResponse := func(err error) {
 		timeout := time.NewTimer(time.Millisecond * 10)
@@ -298,7 +298,7 @@ func (m *Connection) DeviceAuthenticate(ctx context.Context, targetAddress drive
 }
 
 func (m *Connection) DeviceReadProperty(ctx context.Context, targetAddress driverModel.KnxAddress, objectId uint8, propertyId uint8, propertyIndex uint16, numElements uint8) <-chan KnxReadResult {
-	result := make(chan KnxReadResult)
+	result := make(chan KnxReadResult, 1)
 
 	sendResponse := func(value values.PlcValue, numItems uint8, err error) {
 		timeout := time.NewTimer(time.Millisecond * 10)
@@ -382,7 +382,7 @@ func (m *Connection) DeviceReadProperty(ctx context.Context, targetAddress drive
 }
 
 func (m *Connection) DeviceReadPropertyDescriptor(ctx context.Context, targetAddress driverModel.KnxAddress, objectId uint8, propertyId uint8) <-chan KnxReadResult {
-	result := make(chan KnxReadResult)
+	result := make(chan KnxReadResult, 1)
 
 	sendResponse := func(value values.PlcValue, numItems uint8, err error) {
 		timeout := time.NewTimer(time.Millisecond * 10)
@@ -447,7 +447,7 @@ func (m *Connection) DeviceReadPropertyDescriptor(ctx context.Context, targetAdd
 }
 
 func (m *Connection) DeviceReadMemory(ctx context.Context, targetAddress driverModel.KnxAddress, address uint16, numElements uint8, datapointType *driverModel.KnxDatapointType) <-chan KnxReadResult {
-	result := make(chan KnxReadResult)
+	result := make(chan KnxReadResult, 1)
 
 	sendResponse := func(value values.PlcValue, numItems uint8, err error) {
 		timeout := time.NewTimer(time.Millisecond * 10)