Fix: fixed gossip protocol gossip.SendTransaction() now works

main.go

@@ -8,6 +8,7 @@ import (
     "github.com/iotaledger/goshimmer/plugins/gossip"
     "github.com/iotaledger/goshimmer/plugins/gracefulshutdown"
     "github.com/iotaledger/goshimmer/plugins/statusscreen"
+    "github.com/iotaledger/goshimmer/plugins/tangle"
 )
 
 func main() {
@@ -15,6 +16,7 @@ func main() {
         cli.PLUGIN,
         autopeering.PLUGIN,
         gossip.PLUGIN,
+        tangle.PLUGIN,
         analysis.PLUGIN,
         statusscreen.PLUGIN,
         gracefulshutdown.PLUGIN,

packages/errors/errors.go

@@ -107,7 +107,7 @@ func New(message string) *fundamental {
     return &fundamental{
         id:    idCounter,
         msg:   message,
-        stack: callers(),
+        stack: Callers(),
     }
 }
 
@@ -120,7 +120,7 @@ func Errorf(format string, args ...interface{}) IdentifiableError {
     return &fundamental{
         id:    idCounter,
         msg:   fmt.Sprintf(format, args...),
-        stack: callers(),
+        stack: Callers(),
     }
 }
 
@@ -135,7 +135,7 @@ func (f *fundamental) Derive(msg string) *fundamental {
     return &fundamental{
         id:    f.id,
         msg:   msg,
-        stack: callers(),
+        stack: Callers(),
     }
 }
 
@@ -177,7 +177,7 @@ func WithStack(err error) IdentifiableError {
     return &withStack{
         idCounter,
         err,
-        callers(),
+        Callers(),
     }
 }
 
@@ -205,7 +205,7 @@ func (w *withStack) Derive(err error, message string) *withStack {
             cause: err,
             msg:   message,
         },
-        callers(),
+        Callers(),
     }
 }
 
@@ -244,7 +244,7 @@ func Wrap(err error, message string) *withStack {
     return &withStack{
         idCounter,
         err,
-        callers(),
+        Callers(),
     }
 }
 
@@ -265,7 +265,7 @@ func Wrapf(err error, format string, args ...interface{}) IdentifiableError {
     return &withStack{
         idCounter,
         err,
-        callers(),
+        Callers(),
     }
 }
 

packages/errors/stack.go

@@ -160,7 +160,7 @@ func (s *stack) StackTrace() StackTrace {
     return f
 }
 
-func callers() *stack {
+func Callers() *stack {
     const depth = 32
     var pcs [depth]uintptr
     n := runtime.Callers(3, pcs[:])

plugins/gossip/events.go

@@ -78,4 +78,6 @@ func neighborCaller(handler interface{}, params ...interface{}) { handler.(func(
 
 func errorCaller(handler interface{}, params ...interface{}) { handler.(func(errors.IdentifiableError))(params[0].(errors.IdentifiableError)) }
 
+func dataCaller(handler interface{}, params ...interface{}) { handler.(func([]byte))(params[0].([]byte)) }
+
 func transactionCaller(handler interface{}, params ...interface{}) { handler.(func(*transaction.Transaction))(params[0].(*transaction.Transaction)) }

plugins/gossip/neighbors.go

@@ -84,7 +84,7 @@ func manageConnection(plugin *node.Plugin, neighbor *Neighbor) {
                 return
 
             case <-disconnectSignal:
-                break
+                continue
             }
         }
 

plugins/gossip/plugin.go

@@ -12,7 +12,6 @@ func configure(plugin *node.Plugin) {
     configureNeighbors(plugin)
     configureServer(plugin)
     configureSendQueue(plugin)
-    configureTransactionProcessor(plugin)
 
     Events.ReceiveTransaction.Attach(events.NewClosure(func(transaction *transaction.Transaction) {
 
@@ -23,5 +22,4 @@ func run(plugin *node.Plugin) {
     runNeighbors(plugin)
     runServer(plugin)
     runSendQueue(plugin)
-    runTransactionProcessor(plugin)
 }

plugins/gossip/protocol.go

@@ -40,6 +40,7 @@ func newProtocol(conn *network.ManagedConnection) *protocol {
             ReceiveIdentification:     events.NewEvent(identityCaller),
             ReceiveConnectionAccepted: events.NewEvent(events.CallbackCaller),
             ReceiveConnectionRejected: events.NewEvent(events.CallbackCaller),
+            ReceiveTransactionData:    events.NewEvent(dataCaller),
             HandshakeCompleted:        events.NewEvent(events.CallbackCaller),
             Error:                     events.NewEvent(errorCaller),
         },

plugins/gossip/protocol_v1.go

@@ -7,6 +7,7 @@ import (
     "github.com/iotaledger/goshimmer/packages/errors"
     "github.com/iotaledger/goshimmer/packages/events"
     "github.com/iotaledger/goshimmer/packages/identity"
+    "github.com/iotaledger/goshimmer/packages/ternary"
     "github.com/iotaledger/goshimmer/packages/transaction"
     "strconv"
 )
@@ -219,7 +220,7 @@ func newDispatchStateV1(protocol *protocol) *dispatchStateV1 {
 
 func (state *dispatchStateV1) Receive(data []byte, offset int, length int) (int, errors.IdentifiableError) {
     switch data[0] {
-    case 0:
+    case DISPATCH_DROP:
         protocol := state.protocol
 
         protocol.Events.ReceiveConnectionRejected.Trigger()
@@ -228,12 +229,12 @@ func (state *dispatchStateV1) Receive(data []byte, offset int, length int) (int,
 
         protocol.ReceivingState = nil
 
-    case 1:
+    case DISPATCH_TRANSACTION:
         protocol := state.protocol
 
         protocol.ReceivingState = newTransactionStateV1(protocol)
 
-    case 2:
+    case DISPATCH_REQUEST:
         protocol := state.protocol
 
         protocol.ReceivingState = newRequestStateV1(protocol)
@@ -251,6 +252,10 @@ func (state *dispatchStateV1) Send(param interface{}) errors.IdentifiableError {
         case DISPATCH_DROP:
             protocol := state.protocol
 
+            if _, err := protocol.Conn.Write([]byte{DISPATCH_DROP}); err != nil {
+                return ErrSendFailed.Derive(err, "failed to send drop message")
+            }
+
             _ = protocol.Conn.Close()
 
             protocol.SendState = nil
@@ -260,6 +265,10 @@ func (state *dispatchStateV1) Send(param interface{}) errors.IdentifiableError {
         case DISPATCH_TRANSACTION:
             protocol := state.protocol
 
+            if _, err := protocol.Conn.Write([]byte{DISPATCH_TRANSACTION}); err != nil {
+                return ErrSendFailed.Derive(err, "failed to send transaction dispatch byte")
+            }
+
             protocol.SendState = newTransactionStateV1(protocol)
 
             return nil
@@ -267,6 +276,10 @@ func (state *dispatchStateV1) Send(param interface{}) errors.IdentifiableError {
         case DISPATCH_REQUEST:
             protocol := state.protocol
 
+            if _, err := protocol.Conn.Write([]byte{DISPATCH_REQUEST}); err != nil {
+                return ErrSendFailed.Derive(err, "failed to send request dispatch byte")
+            }
+
             protocol.SendState = newTransactionStateV1(protocol)
 
             return nil
@@ -288,8 +301,9 @@ type transactionStateV1 struct {
 
 func newTransactionStateV1(protocol *protocol) *transactionStateV1 {
     return &transactionStateV1{
-        buffer: make([]byte, transaction.MARSHALLED_TOTAL_SIZE),
-        offset: 0,
+        protocol: protocol,
+        buffer:   make([]byte, transaction.MARSHALLED_TOTAL_SIZE / ternary.NUMBER_OF_TRITS_IN_A_BYTE),
+        offset:   0,
     }
 }
 
@@ -297,15 +311,15 @@ func (state *transactionStateV1) Receive(data []byte, offset int, length int) (i
     bytesRead := byteutils.ReadAvailableBytesToBuffer(state.buffer, state.offset, data, offset, length)
 
     state.offset += bytesRead
-    if state.offset == transaction.MARSHALLED_TOTAL_SIZE {
+    if state.offset == transaction.MARSHALLED_TOTAL_SIZE / ternary.NUMBER_OF_TRITS_IN_A_BYTE {
         protocol := state.protocol
 
-        transactionData := make([]byte, transaction.MARSHALLED_TOTAL_SIZE)
+        transactionData := make([]byte, transaction.MARSHALLED_TOTAL_SIZE / ternary.NUMBER_OF_TRITS_IN_A_BYTE)
         copy(transactionData, state.buffer)
 
         protocol.Events.ReceiveTransactionData.Trigger(transactionData)
 
-        go processIncomingTransactionData(transactionData)
+        go ProcessReceivedTransactionData(transactionData)
 
         protocol.ReceivingState = newDispatchStateV1(protocol)
         state.offset = 0
@@ -315,7 +329,19 @@ func (state *transactionStateV1) Receive(data []byte, offset int, length int) (i
 }
 
 func (state *transactionStateV1) Send(param interface{}) errors.IdentifiableError {
-    return nil
+    if tx, ok := param.(*transaction.Transaction); ok {
+        protocol := state.protocol
+
+        if _, err := protocol.Conn.Write(tx.Bytes); err != nil {
+            return ErrSendFailed.Derive(err, "failed to send transaction")
+        }
+
+        protocol.SendState = newDispatchStateV1(protocol)
+
+        return nil
+    }
+
+    return ErrInvalidSendParam.Derive("passed in parameter is not a valid transaction")
 }
 
 // endregion ///////////////////////////////////////////////////////////////////////////////////////////////////////////

plugins/gossip/send_queue.go

@@ -40,10 +40,10 @@ func runSendQueue(plugin *node.Plugin) {
                 for _, neighborQueue := range neighborQueues {
                     select {
                     case neighborQueue.queue <- tx:
-                        return
+                        // log sth
 
                     default:
-                        return
+                        // log sth
                     }
                 }
                 connectedNeighborsMutex.RUnlock()

plugins/gossip/transaction_processor.go

@@ -2,26 +2,25 @@ package gossip
 
 import (
     "github.com/iotaledger/goshimmer/packages/filter"
-    "github.com/iotaledger/goshimmer/packages/node"
     "github.com/iotaledger/goshimmer/packages/transaction"
 )
 
-var transactionFilter = filter.NewByteArrayFilter(TRANSACTION_FILTER_SIZE)
+// region public api ///////////////////////////////////////////////////////////////////////////////////////////////////
 
-func processIncomingTransactionData(transactionData []byte) {
+func ProcessReceivedTransactionData(transactionData []byte) {
     if transactionFilter.Add(transactionData) {
         Events.ReceiveTransaction.Trigger(transaction.FromBytes(transactionData))
     }
 }
 
+// endregion ///////////////////////////////////////////////////////////////////////////////////////////////////////////
 
+// region constants and variables //////////////////////////////////////////////////////////////////////////////////////
 
-func configureTransactionProcessor(plugin *node.Plugin) {
-}
-
-func runTransactionProcessor(plugin *node.Plugin) {
-}
+var transactionFilter = filter.NewByteArrayFilter(TRANSACTION_FILTER_SIZE)
 
 const (
     TRANSACTION_FILTER_SIZE = 5000
 )
+
+// endregion ///////////////////////////////////////////////////////////////////////////////////////////////////////////

plugins/gossip/transaction_processor_test.go

@@ -13,7 +13,7 @@ func BenchmarkProcessSimilarTransactionsFiltered(b *testing.B) {
     b.ResetTimer()
 
     for i := 0; i < b.N; i++ {
-        processIncomingTransactionData(byteArray)
+        ProcessReceivedTransactionData(byteArray)
     }
 }
 

plugins/tangle/plugin.go

 package tangle
 
 import (
+    "fmt"
     "github.com/iotaledger/goshimmer/packages/database"
     "github.com/iotaledger/goshimmer/packages/events"
     "github.com/iotaledger/goshimmer/packages/node"
@@ -8,14 +9,24 @@ import (
     "github.com/iotaledger/goshimmer/plugins/gossip"
 )
 
+// region plugin module setup //////////////////////////////////////////////////////////////////////////////////////////
+
 var PLUGIN = node.NewPlugin("Tangle", configure, run)
 
 func configure(node *node.Plugin) {
+    if db, err := database.Get("transactions"); err != nil {
+        panic(err)
+    } else {
+        transactionDatabase = db
+    }
+
     gossip.Events.ReceiveTransaction.Attach(events.NewClosure(func(transaction *transaction.Transaction) {
         if transactionStoredAlready, err := transactionDatabase.Contains(transaction.Hash.ToBytes()); err != nil {
             panic(err)
         } else {
             if !transactionStoredAlready {
+
+                fmt.Println(transaction.Hash.ToString())
                 // process transaction
             }
         }
@@ -23,15 +34,31 @@ func configure(node *node.Plugin) {
 }
 
 func run(node *node.Plugin) {
+    /*
+    if accountability.OWN_ID.StringIdentifier == "72f84aaee02af4542672cb25aceb9d9e458ef2a3" {
+        go func() {
+            txCounter := 0
+
+            for {
+                txCounter++
+
+                dummyTx := make([]byte, transaction.MARSHALLED_TOTAL_SIZE / ternary.NUMBER_OF_TRITS_IN_A_BYTE)
+                for i := 0; i < 1620; i++ {
+                    dummyTx[i] = byte((i + txCounter) % 128)
+                }
 
+                gossip.SendTransaction(transaction.FromBytes(dummyTx))
+
+                <- time.After(1000 * time.Millisecond)
+            }
+        }()
+    }*/
 }
 
+// endregion ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+// region constants and variables //////////////////////////////////////////////////////////////////////////////////////
+
 var transactionDatabase database.Database
 
-func init() {
-    if db, err := database.Get("transactions"); err != nil {
-        panic(err)
-    } else {
-        transactionDatabase = db
-    }
-}
+// endregion ///////////////////////////////////////////////////////////////////////////////////////////////////////////