refactor(plc4go/spi): abstract bufio.Reader through an interface

plc4go/internal/ads/MessageCodec.go

@@ -20,18 +20,18 @@
 package ads
 
 import (
-	"bufio"
 	"context"
 	"encoding/binary"
-	"github.com/apache/plc4x/plc4go/spi/options"
-	"github.com/rs/zerolog"
 
 	"github.com/apache/plc4x/plc4go/protocols/ads/readwrite/model"
 	"github.com/apache/plc4x/plc4go/spi"
 	"github.com/apache/plc4x/plc4go/spi/default"
+	"github.com/apache/plc4x/plc4go/spi/options"
 	"github.com/apache/plc4x/plc4go/spi/transports"
 	"github.com/apache/plc4x/plc4go/spi/utils"
+
 	"github.com/pkg/errors"
+	"github.com/rs/zerolog"
 )
 
 type MessageCodec struct {
@@ -87,7 +87,7 @@ func (m *MessageCodec) Receive() (spi.Message, error) {
 	transportInstance := m.GetTransportInstance()
 
 	if err := transportInstance.FillBuffer(
-		func(pos uint, currentByte byte, reader *bufio.Reader) bool {
+		func(pos uint, currentByte byte, reader transports.ExtendedReader) bool {
 			numBytesAvailable, err := transportInstance.GetNumBytesAvailableInBuffer()
 			if err != nil {
 				return false
@@ -110,7 +110,7 @@ func (m *MessageCodec) Receive() (spi.Message, error) {
 		packetSize := (uint32(data[5]) << 24) + (uint32(data[4]) << 16) + (uint32(data[3]) << 8) + (uint32(data[2])) + 6
 		if num < packetSize {
 			if err := transportInstance.FillBuffer(
-				func(pos uint, currentByte byte, reader *bufio.Reader) bool {
+				func(pos uint, currentByte byte, reader transports.ExtendedReader) bool {
 					numBytesAvailable, err := transportInstance.GetNumBytesAvailableInBuffer()
 					if err != nil {
 						return false

plc4go/internal/cbus/MessageCodec.go

@@ -20,16 +20,16 @@
 package cbus
 
 import (
-	"bufio"
 	"context"
+	"sync"
+	"sync/atomic"
+	"time"
+
 	readWriteModel "github.com/apache/plc4x/plc4go/protocols/cbus/readwrite/model"
 	"github.com/apache/plc4x/plc4go/spi"
 	"github.com/apache/plc4x/plc4go/spi/default"
 	"github.com/apache/plc4x/plc4go/spi/options"
 	"github.com/apache/plc4x/plc4go/spi/transports"
-	"sync"
-	"sync/atomic"
-	"time"
 
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
@@ -129,10 +129,13 @@ func (m *MessageCodec) Send(message spi.Message) error {
 func (m *MessageCodec) Receive() (spi.Message, error) {
 	m.log.Trace().Msg("Receive")
 	ti := m.GetTransportInstance()
+	if !ti.IsConnected() {
+		return nil, errors.New("Transport instance not connected")
+	}
 	confirmation := false
 	// Fill the buffer
 	{
-		if err := ti.FillBuffer(func(pos uint, currentByte byte, reader *bufio.Reader) bool {
+		if err := ti.FillBuffer(func(pos uint, currentByte byte, reader transports.ExtendedReader) bool {
 			m.log.Trace().Uint8("byte", currentByte).Msg("current byte")
 			switch currentByte {
 			case

plc4go/spi/transports/ExtendedReader.go

@@ -0,0 +1,31 @@
+/*
+ * 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 transports
+
+import "io"
+
+type ExtendedReader interface {
+	io.Reader
+	io.ByteReader
+	// Peek returns the next n bytes without advancing the reader.
+	Peek(int) ([]byte, error)
+	// Buffered returns the number of bytes that can be read from the current buffer.
+	Buffered() int
+}

plc4go/spi/transports/TransportInstance.go

@@ -20,7 +20,6 @@
 package transports
 
 import (
-	"bufio"
 	"context"
 	"fmt"
 )
@@ -34,7 +33,7 @@ type TransportInstance interface {
 	IsConnected() bool
 
 	// FillBuffer fills the buffer `until` false (Useful in conjunction if you want GetNumBytesAvailableInBuffer)
-	FillBuffer(until func(pos uint, currentByte byte, reader *bufio.Reader) bool) error
+	FillBuffer(until func(pos uint, currentByte byte, reader ExtendedReader) bool) error
 	// GetNumBytesAvailableInBuffer returns the bytes currently available in buffer (!!!Careful: if you looking for a termination you have to use FillBuffer)
 	GetNumBytesAvailableInBuffer() (uint32, error)
 	PeekReadableBytes(numBytes uint32) ([]byte, error)