diff --git a/go.mod b/go.mod
index 01ea03c9b72bffc0856fe3fbb96b5881263b587f..aa3d705bcb4ec272cf43a25dde3e24f396368d2a 100644
--- a/go.mod
+++ b/go.mod
@@ -10,7 +10,7 @@ require (
github.com/googollee/go-engine.io v1.4.3-0.20190924125625-798118fc0dd2
github.com/googollee/go-socket.io v1.4.3-0.20191204093753-683f8725b6d0
github.com/gorilla/websocket v1.4.1
- github.com/iotaledger/hive.go v0.0.0-20200304073310-d71239623d46
+ github.com/iotaledger/hive.go v0.0.0-20200305153009-d3408b2be05b
github.com/iotaledger/iota.go v1.0.0-beta.14
github.com/labstack/echo v3.3.10+incompatible
github.com/labstack/gommon v0.3.0 // indirect
diff --git a/go.sum b/go.sum
index db2613b1604ec816f1375ac2c04d29a0ac5c0966..d573f25b2ba5a39ed6bd477f703fe5b90044a923 100644
--- a/go.sum
+++ b/go.sum
@@ -136,6 +136,8 @@ github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpO
github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
github.com/iotaledger/hive.go v0.0.0-20200304073310-d71239623d46 h1:4v/lfoR5CiJKsG8HUuYeKbds+DPvpxxxSrKZYS7Qhrc=
github.com/iotaledger/hive.go v0.0.0-20200304073310-d71239623d46/go.mod h1:0LQvxKmfU4bcQcjYIAq3PRfsA5584U0AioAAas6/QU8=
+github.com/iotaledger/hive.go v0.0.0-20200305153009-d3408b2be05b h1:TxHm2TTOG4Xgxdnp5qZsgXKCctnxdDeQgqT4NwPJPQ0=
+github.com/iotaledger/hive.go v0.0.0-20200305153009-d3408b2be05b/go.mod h1:0LQvxKmfU4bcQcjYIAq3PRfsA5584U0AioAAas6/QU8=
github.com/iotaledger/iota.go v1.0.0-beta.9/go.mod h1:F6WBmYd98mVjAmmPVYhnxg8NNIWCjjH8VWT9qvv3Rc8=
github.com/iotaledger/iota.go v1.0.0-beta.14 h1:Oeb28MfBuJEeXcGrLhTCJFtbsnc8y1u7xidsAmiOD5A=
github.com/iotaledger/iota.go v1.0.0-beta.14/go.mod h1:F6WBmYd98mVjAmmPVYhnxg8NNIWCjjH8VWT9qvv3Rc8=
diff --git a/packages/binary/tangle/tangle_test.go b/packages/binary/tangle/tangle_test.go
index 2ef3cdc18f518a06e82460607857967c907ce5bd..f0f8c60ef5a7660306818fc83d7d0bfdc821fd66 100644
--- a/packages/binary/tangle/tangle_test.go
+++ b/packages/binary/tangle/tangle_test.go
@@ -2,18 +2,20 @@ package tangle
import (
"fmt"
+ "io/ioutil"
+ "os"
"testing"
"time"
"github.com/dgraph-io/badger/v2"
"github.com/iotaledger/hive.go/events"
-
- "github.com/iotaledger/hive.go/database"
+ "github.com/stretchr/testify/require"
"github.com/iotaledger/goshimmer/packages/binary/identity"
"github.com/iotaledger/goshimmer/packages/binary/tangle/model/transaction"
"github.com/iotaledger/goshimmer/packages/binary/tangle/model/transaction/payload/data"
"github.com/iotaledger/goshimmer/packages/binary/tangle/model/transactionmetadata"
+ "github.com/iotaledger/goshimmer/packages/database"
"github.com/iotaledger/goshimmer/plugins/config"
)
@@ -21,11 +23,13 @@ var testDatabase *badger.DB
var _ = config.PLUGIN
-func init() {
- testDatabase = database.GetBadgerInstance()
-}
-
func BenchmarkTangle_AttachTransaction(b *testing.B) {
+ dir, err := ioutil.TempDir("", b.Name())
+ require.NoError(b, err)
+ defer os.Remove(dir)
+ // use the tempdir for the database
+ config.Node.Set(database.CFG_DIRECTORY, dir)
+
tangle := New(testDatabase, []byte("TEST_BINARY_TANGLE"))
if err := tangle.Prune(); err != nil {
b.Error(err)
@@ -51,7 +55,13 @@ func BenchmarkTangle_AttachTransaction(b *testing.B) {
}
func TestTangle_AttachTransaction(t *testing.T) {
- tangle := New(testDatabase, []byte("TEST_BINARY_TANGLE"))
+ dir, err := ioutil.TempDir("", t.Name())
+ require.NoError(t, err)
+ defer os.Remove(dir)
+ // use the tempdir for the database
+ config.Node.Set(database.CFG_DIRECTORY, dir)
+
+ tangle := New(database.GetBadgerInstance(), []byte("TEST_BINARY_TANGLE"))
if err := tangle.Prune(); err != nil {
t.Error(err)
diff --git a/packages/database/database.go b/packages/database/database.go
index a815d4274231ccf5b0cf2740c6234ca768fc0c85..809ef4595fcc6f75b1ca82bc72a22e7d3b5b81f5 100644
--- a/packages/database/database.go
+++ b/packages/database/database.go
@@ -1,3 +1,4 @@
+// Wrapper for hive.go/database package. Only use this instead of the hive.go package.
package database
import (
diff --git a/packages/database/mapdb/mapdb.go b/packages/database/mapdb/mapdb.go
deleted file mode 100644
index e9627130049cc2b2a7bcba0787c2c30036d420ab..0000000000000000000000000000000000000000
--- a/packages/database/mapdb/mapdb.go
+++ /dev/null
@@ -1,220 +0,0 @@
-// Package mapdb provides a map implementation of a key value database.
-// It offers a lightweight drop-in replacement of hive.go/database for tests or in simulations
-// where more than one instance is required.
-package mapdb
-
-import (
- "strings"
- "sync"
-
- "github.com/iotaledger/hive.go/database"
- "github.com/iotaledger/hive.go/typeutils"
-)
-
-// MapDB is a simple implementation of DB using a map.
-type MapDB struct {
- mu sync.RWMutex
- m map[string]mapEntry
-}
-
-type mapEntry struct {
- value []byte
- meta byte
-}
-
-// NewMapDB creates a database.Database implementation purely based on a go map.
-// MapDB does not support TTL.
-func NewMapDB() *MapDB {
- return &MapDB{
- m: make(map[string]mapEntry),
- }
-}
-
-func (db *MapDB) Contains(key database.Key) (contains bool, err error) {
- db.mu.RLock()
- defer db.mu.RUnlock()
-
- _, contains = db.m[typeutils.BytesToString(key)]
- return
-}
-
-func (db *MapDB) Get(key database.Key) (entry database.Entry, err error) {
- db.mu.RLock()
- defer db.mu.RUnlock()
-
- ent, contains := db.m[typeutils.BytesToString(key)]
- if !contains {
- err = database.ErrKeyNotFound
- return
- }
- entry.Key = key
- entry.Value = append([]byte{}, ent.value...)
- entry.Meta = ent.meta
- return
-}
-
-func (db *MapDB) GetKeyOnly(key database.Key) (entry database.KeyOnlyEntry, err error) {
- db.mu.RLock()
- defer db.mu.RUnlock()
-
- ent, contains := db.m[typeutils.BytesToString(key)]
- if !contains {
- err = database.ErrKeyNotFound
- return
- }
- entry.Key = key
- entry.Meta = ent.meta
- return
-}
-
-func (db *MapDB) Set(entry database.Entry) error {
- db.mu.Lock()
- defer db.mu.Unlock()
-
- db.m[typeutils.BytesToString(entry.Key)] = mapEntry{
- value: append([]byte{}, entry.Value...),
- meta: entry.Meta,
- }
- return nil
-}
-
-func (db *MapDB) Delete(key database.Key) error {
- db.mu.Lock()
- defer db.mu.Unlock()
-
- delete(db.m, typeutils.BytesToString(key))
- return nil
-}
-
-func (db *MapDB) DeletePrefix(keyPrefix database.KeyPrefix) error {
- db.mu.Lock()
- defer db.mu.Unlock()
-
- prefix := typeutils.BytesToString(keyPrefix)
- for key := range db.m {
- if strings.HasPrefix(key, prefix) {
- delete(db.m, key)
- }
- }
- return nil
-}
-
-func (db *MapDB) ForEach(consume func(entry database.Entry) bool) error {
- db.mu.RLock()
- defer db.mu.RUnlock()
-
- for key, ent := range db.m {
- entry := database.Entry{
- Key: []byte(key),
- Value: append([]byte{}, ent.value...),
- Meta: ent.meta,
- }
- if consume(entry) {
- break
- }
- }
- return nil
-}
-
-func (db *MapDB) ForEachKeyOnly(consume func(entry database.KeyOnlyEntry) bool) error {
- db.mu.RLock()
- defer db.mu.RUnlock()
-
- for key, ent := range db.m {
- entry := database.KeyOnlyEntry{
- Key: []byte(key),
- Meta: ent.meta,
- }
- if consume(entry) {
- break
- }
- }
- return nil
-}
-
-func (db *MapDB) ForEachPrefix(keyPrefix database.KeyPrefix, consume func(entry database.Entry) (stop bool)) error {
- db.mu.RLock()
- defer db.mu.RUnlock()
-
- prefix := typeutils.BytesToString(keyPrefix)
- for key, ent := range db.m {
- if strings.HasPrefix(key, prefix) {
- entry := database.Entry{
- Key: []byte(strings.TrimPrefix(key, prefix)),
- Value: append([]byte{}, ent.value...),
- Meta: ent.meta,
- }
- if consume(entry) {
- break
- }
- }
- }
- return nil
-}
-
-func (db *MapDB) ForEachPrefixKeyOnly(keyPrefix database.KeyPrefix, consume func(entry database.KeyOnlyEntry) (stop bool)) error {
- db.mu.RLock()
- defer db.mu.RUnlock()
-
- prefix := typeutils.BytesToString(keyPrefix)
- for key, ent := range db.m {
- if strings.HasPrefix(key, prefix) {
- entry := database.KeyOnlyEntry{
- Key: []byte(strings.TrimPrefix(key, prefix)),
- Meta: ent.meta,
- }
- if consume(entry) {
- break
- }
- }
- }
- return nil
-}
-
-func (db *MapDB) StreamForEach(consume func(entry database.Entry) error) (err error) {
- _ = db.ForEach(func(entry database.Entry) bool {
- err = consume(entry)
- return err != nil
- })
- return
-}
-
-func (db *MapDB) StreamForEachKeyOnly(consume func(entry database.KeyOnlyEntry) error) (err error) {
- _ = db.ForEachKeyOnly(func(entry database.KeyOnlyEntry) bool {
- err = consume(entry)
- return err != nil
- })
- return
-}
-
-func (db *MapDB) StreamForEachPrefix(keyPrefix database.KeyPrefix, consume func(entry database.Entry) error) (err error) {
- _ = db.ForEachPrefix(keyPrefix, func(entry database.Entry) bool {
- err = consume(entry)
- return err != nil
- })
- return
-}
-
-func (db *MapDB) StreamForEachPrefixKeyOnly(keyPrefix database.KeyPrefix, consume func(database.KeyOnlyEntry) error) (err error) {
- _ = db.ForEachPrefixKeyOnly(keyPrefix, func(entry database.KeyOnlyEntry) bool {
- err = consume(entry)
- return err != nil
- })
- return
-}
-
-func (db *MapDB) Apply(set []database.Entry, del []database.Key) error {
- db.mu.Lock()
- defer db.mu.Unlock()
-
- for _, entry := range set {
- db.m[typeutils.BytesToString(entry.Key)] = mapEntry{
- value: append([]byte{}, entry.Value...),
- meta: entry.Meta,
- }
- }
- for _, key := range del {
- delete(db.m, typeutils.BytesToString(key))
- }
- return nil
-}
diff --git a/packages/gossip/manager_test.go b/packages/gossip/manager_test.go
index 3d6b89b573a637be30fdd42ff42e794ce9910d11..afb6ee59d707b2f53ff0e0f4ad070bb67dce170b 100644
--- a/packages/gossip/manager_test.go
+++ b/packages/gossip/manager_test.go
@@ -8,11 +8,11 @@ import (
"github.com/golang/protobuf/proto"
"github.com/iotaledger/goshimmer/packages/binary/tangle/model/transaction"
- "github.com/iotaledger/goshimmer/packages/database/mapdb"
pb "github.com/iotaledger/goshimmer/packages/gossip/proto"
"github.com/iotaledger/goshimmer/packages/gossip/server"
"github.com/iotaledger/hive.go/autopeering/peer"
"github.com/iotaledger/hive.go/autopeering/peer/service"
+ "github.com/iotaledger/hive.go/database/mapdb"
"github.com/iotaledger/hive.go/events"
"github.com/iotaledger/hive.go/logger"
"github.com/stretchr/testify/assert"
diff --git a/packages/gossip/neighbor.go b/packages/gossip/neighbor.go
index a529aab2af7ede70439dd696a3d15f2e7a1cd8c4..4e5ced9f7d12266a9cb54be38ac5ba55a69a2a4c 100644
--- a/packages/gossip/neighbor.go
+++ b/packages/gossip/neighbor.go
@@ -7,10 +7,10 @@ import (
"strings"
"sync"
- "github.com/iotaledger/goshimmer/packages/netutil"
- "github.com/iotaledger/goshimmer/packages/netutil/buffconn"
"github.com/iotaledger/hive.go/autopeering/peer"
"github.com/iotaledger/hive.go/logger"
+ "github.com/iotaledger/hive.go/netutil"
+ "github.com/iotaledger/hive.go/netutil/buffconn"
"go.uber.org/atomic"
)
diff --git a/packages/gossip/server/server.go b/packages/gossip/server/server.go
index 6185df511f9b6c0f794ac532611b1bb3d5b8c249..6fe98d5363f6eb4ea87e09c13fb54f8d9fc88b67 100644
--- a/packages/gossip/server/server.go
+++ b/packages/gossip/server/server.go
@@ -12,11 +12,11 @@ import (
"time"
"github.com/golang/protobuf/proto"
- "github.com/iotaledger/goshimmer/packages/netutil"
"github.com/iotaledger/hive.go/autopeering/peer"
"github.com/iotaledger/hive.go/autopeering/peer/service"
pb "github.com/iotaledger/hive.go/autopeering/server/proto"
"github.com/iotaledger/hive.go/backoff"
+ "github.com/iotaledger/hive.go/netutil"
"go.uber.org/zap"
)
diff --git a/packages/gossip/server/server_test.go b/packages/gossip/server/server_test.go
index d3e954184da75295687c3710db34a9fd317aa756..602968e3dda75974e86cf43c7964940f85d08ae2 100644
--- a/packages/gossip/server/server_test.go
+++ b/packages/gossip/server/server_test.go
@@ -6,9 +6,9 @@ import (
"testing"
"time"
- "github.com/iotaledger/goshimmer/packages/database/mapdb"
"github.com/iotaledger/hive.go/autopeering/peer"
"github.com/iotaledger/hive.go/autopeering/peer/service"
+ "github.com/iotaledger/hive.go/database/mapdb"
"github.com/iotaledger/hive.go/logger"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
diff --git a/packages/netutil/buffconn/buffconn.go b/packages/netutil/buffconn/buffconn.go
deleted file mode 100644
index c1ad9eaa28544f3091a1a3788798a4d2cdf6ae83..0000000000000000000000000000000000000000
--- a/packages/netutil/buffconn/buffconn.go
+++ /dev/null
@@ -1,190 +0,0 @@
-package buffconn
-
-import (
- "encoding/binary"
- "errors"
- "fmt"
- "net"
- "sync"
- "time"
-
- "github.com/iotaledger/hive.go/events"
- "go.uber.org/atomic"
-)
-
-const (
- // MaxMessageSize is the maximum message size in bytes.
- MaxMessageSize = 4096
- // IOTimeout specifies the timeout for sending and receiving multi packet messages.
- IOTimeout = 4 * time.Second
-
- headerSize = 4 // size of the header: uint32
-)
-
-// Errors returned by the BufferedConnection.
-var (
- ErrInvalidHeader = errors.New("invalid message header")
- ErrInsufficientBuffer = errors.New("insufficient buffer")
-)
-
-// BufferedConnectionEvents contains all the events that are triggered during the peer discovery.
-type BufferedConnectionEvents struct {
- ReceiveMessage *events.Event
- Close *events.Event
-}
-
-// BufferedConnection is a wrapper for sending and reading messages with a buffer.
-type BufferedConnection struct {
- Events BufferedConnectionEvents
-
- conn net.Conn
- incomingHeaderBuffer []byte
- closeOnce sync.Once
-
- bytesRead *atomic.Uint32
- bytesWritten *atomic.Uint32
-}
-
-// NewBufferedConnection creates a new BufferedConnection from a net.Conn.
-func NewBufferedConnection(conn net.Conn) *BufferedConnection {
- return &BufferedConnection{
- Events: BufferedConnectionEvents{
- ReceiveMessage: events.NewEvent(events.ByteSliceCaller),
- Close: events.NewEvent(events.CallbackCaller),
- },
- conn: conn,
- incomingHeaderBuffer: make([]byte, headerSize),
- bytesRead: atomic.NewUint32(0),
- bytesWritten: atomic.NewUint32(0),
- }
-}
-
-// Close closes the connection.
-// Any blocked Read or Write operations will be unblocked and return errors.
-func (c *BufferedConnection) Close() (err error) {
- c.closeOnce.Do(func() {
- err = c.conn.Close()
- // close in separate go routine to avoid deadlocks
- go c.Events.Close.Trigger()
- })
- return err
-}
-
-// LocalAddr returns the local network address.
-func (c *BufferedConnection) LocalAddr() net.Addr {
- return c.conn.LocalAddr()
-}
-
-// RemoteAddr returns the remote network address.
-func (c *BufferedConnection) RemoteAddr() net.Addr {
- return c.conn.RemoteAddr()
-}
-
-// BytesRead returns the total number of bytes read.
-func (c *BufferedConnection) BytesRead() uint32 {
- return c.bytesRead.Load()
-}
-
-// BytesWritten returns the total number of bytes written.
-func (c *BufferedConnection) BytesWritten() uint32 {
- return c.bytesWritten.Load()
-}
-
-// Read starts reading on the connection, it only returns when an error occurred or when Close has been called.
-// If a complete message has been received and ReceiveMessage event is triggered with its complete payload.
-// If read leads to an error, the loop will be stopped and that error returned.
-func (c *BufferedConnection) Read() error {
- buffer := make([]byte, MaxMessageSize)
-
- for {
- n, err := c.readMessage(buffer)
- if err != nil {
- return err
- }
- if n > 0 {
- c.Events.ReceiveMessage.Trigger(buffer[:n])
- }
- }
-}
-
-// Write sends a stream of bytes as messages.
-// Each array of bytes you pass in will be pre-pended with it's size. If the
-// connection isn't open you will receive an error. If not all bytes can be
-// written, Write will keep trying until the full message is delivered, or the
-// connection is broken.
-func (c *BufferedConnection) Write(msg []byte) (int, error) {
- if l := len(msg); l > MaxMessageSize {
- panic(fmt.Sprintf("invalid message length: %d", l))
- }
-
- buffer := append(newHeader(len(msg)), msg...)
-
- if err := c.conn.SetWriteDeadline(time.Now().Add(IOTimeout)); err != nil {
- return 0, fmt.Errorf("error while setting timeout: %w", err)
- }
-
- toWrite := len(buffer)
- for bytesWritten := 0; bytesWritten < toWrite; {
- n, err := c.conn.Write(buffer[bytesWritten:])
- bytesWritten += n
- c.bytesWritten.Add(uint32(n))
- if err != nil {
- return bytesWritten, err
- }
- }
- return toWrite - headerSize, nil
-}
-
-func (c *BufferedConnection) read(buffer []byte) (int, error) {
- toRead := len(buffer)
- for bytesRead := 0; bytesRead < toRead; {
- n, err := c.conn.Read(buffer[bytesRead:])
- bytesRead += n
- c.bytesRead.Add(uint32(n))
- if err != nil {
- return bytesRead, err
- }
- }
- return toRead, nil
-}
-
-func (c *BufferedConnection) readMessage(buffer []byte) (int, error) {
- if err := c.conn.SetReadDeadline(time.Time{}); err != nil {
- return 0, fmt.Errorf("error while unsetting timeout: %w", err)
- }
- _, err := c.read(c.incomingHeaderBuffer)
- if err != nil {
- return 0, err
- }
-
- msgLength, err := parseHeader(c.incomingHeaderBuffer)
- if err != nil {
- return 0, err
- }
- if msgLength > len(buffer) {
- return 0, ErrInsufficientBuffer
- }
-
- if err := c.conn.SetReadDeadline(time.Now().Add(IOTimeout)); err != nil {
- return 0, fmt.Errorf("error while setting timeout: %w", err)
- }
- return c.read(buffer[:msgLength])
-}
-
-func newHeader(msgLength int) []byte {
- // the header only consists of the message length
- header := make([]byte, headerSize)
- binary.BigEndian.PutUint32(header, uint32(msgLength))
- return header
-}
-
-func parseHeader(header []byte) (int, error) {
- if len(header) != headerSize {
- return 0, ErrInvalidHeader
- }
- msgLength := int(binary.BigEndian.Uint32(header))
- if msgLength > MaxMessageSize {
- return 0, ErrInvalidHeader
- }
- return msgLength, nil
-}
diff --git a/packages/netutil/buffconn/buffconn_test.go b/packages/netutil/buffconn/buffconn_test.go
deleted file mode 100644
index e24506b433b79c298c391acfde015d412b57f629..0000000000000000000000000000000000000000
--- a/packages/netutil/buffconn/buffconn_test.go
+++ /dev/null
@@ -1,136 +0,0 @@
-package buffconn
-
-import (
- "errors"
- "io"
- "net"
- "sync"
- "testing"
- "time"
-
- "github.com/iotaledger/hive.go/events"
- "github.com/stretchr/testify/assert"
- "github.com/stretchr/testify/require"
-)
-
-const graceTime = 10 * time.Millisecond
-
-var testMsg = []byte("test")
-
-func TestBufferedConnection(t *testing.T) {
- t.Run("Close", func(t *testing.T) {
- conn1, conn2 := net.Pipe()
- buffConn1 := NewBufferedConnection(conn1)
- defer buffConn1.Close()
- buffConn2 := NewBufferedConnection(conn2)
- defer buffConn2.Close()
-
- var wg sync.WaitGroup
- wg.Add(2)
- buffConn2.Events.Close.Attach(events.NewClosure(func() { wg.Done() }))
- go func() {
- err := buffConn2.Read()
- assert.True(t, errors.Is(err, io.ErrClosedPipe), "unexpected error: %s", err)
- require.NoError(t, buffConn2.Close())
- wg.Done()
- }()
-
- err := buffConn1.Close()
- require.NoError(t, err)
- wg.Wait()
- })
-
- t.Run("Write", func(t *testing.T) {
- conn1, conn2 := net.Pipe()
- buffConn1 := NewBufferedConnection(conn1)
- defer buffConn1.Close()
- buffConn2 := NewBufferedConnection(conn2)
- defer buffConn2.Close()
-
- go func() {
- _ = buffConn2.Read()
- }()
-
- n, err := buffConn1.Write(testMsg)
- require.NoError(t, err)
- assert.EqualValues(t, len(testMsg), n)
- })
-
- t.Run("ReceiveMessage", func(t *testing.T) {
- conn1, conn2 := net.Pipe()
- buffConn1 := NewBufferedConnection(conn1)
- defer buffConn1.Close()
- buffConn2 := NewBufferedConnection(conn2)
- defer buffConn2.Close()
-
- var wg sync.WaitGroup
- wg.Add(2)
- buffConn2.Events.ReceiveMessage.Attach(events.NewClosure(func(data []byte) {
- assert.EqualValues(t, testMsg, data)
- wg.Done()
- }))
- go func() {
- err := buffConn2.Read()
- assert.True(t, errors.Is(err, io.EOF), "unexpected error: %s", err)
- wg.Done()
- }()
-
- n, err := buffConn1.Write(testMsg)
- require.NoError(t, err)
- assert.EqualValues(t, len(testMsg), n)
-
- time.Sleep(graceTime)
-
- err = buffConn1.Close()
- require.NoError(t, err)
- wg.Wait()
- })
-
- t.Run("ReceiveMany", func(t *testing.T) {
- conn1, conn2 := net.Pipe()
- buffConn1 := NewBufferedConnection(conn1)
- defer buffConn1.Close()
- buffConn2 := NewBufferedConnection(conn2)
- defer buffConn2.Close()
-
- const numWrites = 3
-
- var wg sync.WaitGroup
- wg.Add(numWrites)
- buffConn2.Events.ReceiveMessage.Attach(events.NewClosure(func(data []byte) {
- assert.Equal(t, testMsg, data)
- wg.Done()
- }))
- go func() {
- _ = buffConn2.Read()
- }()
-
- for i := 1; i <= numWrites; i++ {
- _, err := buffConn1.Write(testMsg)
- require.NoError(t, err)
- if i < numWrites {
- time.Sleep(IOTimeout + graceTime)
- }
- }
- wg.Wait()
- })
-
- t.Run("InvalidHeader", func(t *testing.T) {
- conn1, conn2 := net.Pipe()
- buffConn1 := NewBufferedConnection(conn1)
- defer buffConn1.Close()
- defer conn2.Close()
-
- var wg sync.WaitGroup
- wg.Add(1)
- go func() {
- err := buffConn1.Read()
- assert.True(t, errors.Is(err, ErrInvalidHeader), "unexpected error: %s", err)
- wg.Done()
- }()
-
- _, err := conn2.Write([]byte{0xff, 0xff, 0xff, 0xff})
- require.NoError(t, err)
- wg.Wait()
- })
-}
diff --git a/packages/netutil/netutil.go b/packages/netutil/netutil.go
deleted file mode 100644
index b9c4932c95410068dcfc522fe583a914dc1ee972..0000000000000000000000000000000000000000
--- a/packages/netutil/netutil.go
+++ /dev/null
@@ -1,104 +0,0 @@
-// Package netutil provides utility functions extending the stdnet package.
-package netutil
-
-import (
- "bytes"
- "encoding/binary"
- "errors"
- "fmt"
- "io/ioutil"
- "math/rand"
- "net"
- "net/http"
- "time"
-)
-
-var (
- errInvalidData = errors.New("invalid data received")
-)
-
-// IsIPv4 returns true if ip is an IPv4 address.
-func IsIPv4(ip net.IP) bool {
- return ip.To4() != nil
-}
-
-// GetPublicIP queries the ipify API for the public IP address.
-func GetPublicIP(preferIPv6 bool) (net.IP, error) {
- var url string
- if preferIPv6 {
- url = "https://api6.ipify.org"
- } else {
- url = "https://api.ipify.org"
- }
- resp, err := http.Get(url)
- if err != nil {
- return nil, fmt.Errorf("get failed: %w", err)
- }
- defer resp.Body.Close()
-
- body, err := ioutil.ReadAll(resp.Body)
- if err != nil {
- return nil, fmt.Errorf("read failed: %w", err)
- }
-
- // the body only consists of the ip address
- ip := net.ParseIP(string(body))
- if ip == nil {
- return nil, fmt.Errorf("not an IP: %s", body)
- }
- return ip, nil
-}
-
-// IsTemporaryError checks whether the given error should be considered temporary.
-func IsTemporaryError(err error) bool {
- tempErr, ok := err.(interface {
- Temporary() bool
- })
- return ok && tempErr.Temporary()
-}
-
-// CheckUDP checks whether data send to remote is received at local, otherwise an error is returned.
-// If checkAddress is set, it checks whether the IP address that was on the packet matches remote.
-// If checkPort is set, it checks whether the port that was on the packet matches remote.
-func CheckUDP(local, remote *net.UDPAddr, checkAddress bool, checkPort bool) error {
- conn, err := net.ListenUDP("udp", local)
- if err != nil {
- return fmt.Errorf("listen failed: %w", err)
- }
- defer conn.Close()
-
- nonce := generateNonce()
- _, err = conn.WriteTo(nonce, remote)
- if err != nil {
- return fmt.Errorf("write failed: %w", err)
- }
-
- err = conn.SetReadDeadline(time.Now().Add(2 * time.Second))
- if err != nil {
- return fmt.Errorf("set timeout failed: %w", err)
- }
-
- p := make([]byte, len(nonce)+1)
- n, from, err := conn.ReadFrom(p)
- if err != nil {
- return fmt.Errorf("read failed: %w", err)
- }
- if n != len(nonce) || !bytes.Equal(p[:n], nonce) {
- return errInvalidData
- }
- udpAddr := from.(*net.UDPAddr)
- if checkAddress && !udpAddr.IP.Equal(remote.IP) {
- return fmt.Errorf("IP changed: %s", udpAddr.IP)
- }
- if checkPort && udpAddr.Port != remote.Port {
- return fmt.Errorf("port changed: %d", udpAddr.Port)
- }
-
- return nil
-}
-
-func generateNonce() []byte {
- b := make([]byte, 8)
- binary.BigEndian.PutUint64(b, rand.Uint64())
- return b
-}
diff --git a/packages/netutil/netutil_test.go b/packages/netutil/netutil_test.go
deleted file mode 100644
index 99fca2cb3dd85242dd2e7cc26f4d97fddf7c68c1..0000000000000000000000000000000000000000
--- a/packages/netutil/netutil_test.go
+++ /dev/null
@@ -1,71 +0,0 @@
-package netutil
-
-import (
- "errors"
- "fmt"
- "net"
- "testing"
-
- "github.com/stretchr/testify/assert"
- "github.com/stretchr/testify/require"
-)
-
-func TestIsIPv4(t *testing.T) {
- tests := []struct {
- in net.IP
- out bool
- }{
- {nil, false},
- {net.IPv4zero, true},
- {net.IPv6zero, false},
- {net.ParseIP("127.0.0.1"), true},
- {net.IPv6loopback, false},
- {net.ParseIP("8.8.8.8"), true},
- {net.ParseIP("2001:4860:4860::8888"), false},
- }
- for _, tt := range tests {
- t.Run(fmt.Sprintf("%v", tt.in), func(t *testing.T) {
- assert.Equal(t, IsIPv4(tt.in), tt.out)
- })
- }
-}
-
-func TestIsTemporaryError(t *testing.T) {
- tests := []struct {
- in error
- out bool
- }{
- {nil, false},
- {errors.New("errorString"), false},
- }
- for _, tt := range tests {
- t.Run(fmt.Sprintf("%v", tt.in), func(t *testing.T) {
- assert.Equal(t, IsTemporaryError(tt.in), tt.out)
- })
- }
-}
-
-func TestCheckUDP(t *testing.T) {
- local, err := getLocalUDPAddr()
- require.NoError(t, err)
- assert.NoError(t, CheckUDP(local, local, true, true))
-
- invalid := &net.UDPAddr{
- IP: local.IP,
- Port: local.Port - 1,
- Zone: local.Zone,
- }
- assert.Error(t, CheckUDP(local, invalid, false, false))
-}
-
-func getLocalUDPAddr() (*net.UDPAddr, error) {
- addr, err := net.ResolveUDPAddr("udp", "127.0.0.1:0")
- if err != nil {
- return nil, err
- }
- conn, err := net.ListenUDP("udp", addr)
- if err != nil {
- return nil, err
- }
- return conn.LocalAddr().(*net.UDPAddr), conn.Close()
-}
diff --git a/plugins/autopeering/local/local.go b/plugins/autopeering/local/local.go
index ef68c92473b9450fe29545f99d55bfa418122ee1..1586cbff8862e41177d918c4b18b6409800aada3 100644
--- a/plugins/autopeering/local/local.go
+++ b/plugins/autopeering/local/local.go
@@ -11,9 +11,9 @@ import (
"github.com/iotaledger/hive.go/autopeering/peer"
"github.com/iotaledger/hive.go/autopeering/peer/service"
"github.com/iotaledger/hive.go/logger"
+ "github.com/iotaledger/hive.go/netutil"
"github.com/iotaledger/goshimmer/packages/database"
- "github.com/iotaledger/goshimmer/packages/netutil"
"github.com/iotaledger/goshimmer/plugins/config"
)
diff --git a/plugins/portcheck/plugin.go b/plugins/portcheck/plugin.go
index b87d29b53f6eaa49c1700a3b29336b07b4425f95..09f25947361b5145329dbe9f8087ea7a99e45481 100644
--- a/plugins/portcheck/plugin.go
+++ b/plugins/portcheck/plugin.go
@@ -5,7 +5,6 @@ import (
"sync"
"github.com/iotaledger/goshimmer/packages/gossip/server"
- "github.com/iotaledger/goshimmer/packages/netutil"
"github.com/iotaledger/goshimmer/plugins/autopeering"
"github.com/iotaledger/goshimmer/plugins/autopeering/local"
"github.com/iotaledger/goshimmer/plugins/banner"
@@ -14,6 +13,7 @@ import (
"github.com/iotaledger/hive.go/autopeering/peer/service"
"github.com/iotaledger/hive.go/logger"
+ "github.com/iotaledger/hive.go/netutil"
"github.com/iotaledger/hive.go/node"
)