-
Notifications
You must be signed in to change notification settings - Fork 0
/
manager.go
113 lines (89 loc) · 3.85 KB
/
manager.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
package protocol
import (
"fmt"
"sync"
"sync/atomic"
"github.com/karlsend/PYVERT/testfork/karlsend/app/protocol/common"
"github.com/pkg/errors"
"github.com/karlsend/PYVERT/testfork/karlsend/domain"
"github.com/karlsend/PYVERT/testfork/karlsend/domain/consensus/model/externalapi"
"github.com/karlsend/PYVERT/testfork/karlsend/app/protocol/flowcontext"
peerpkg "github.com/karlsend/PYVERT/testfork/karlsend/app/protocol/peer"
"github.com/karlsend/PYVERT/testfork/karlsend/infrastructure/config"
"github.com/karlsend/PYVERT/testfork/karlsend/infrastructure/network/addressmanager"
"github.com/karlsend/PYVERT/testfork/karlsend/infrastructure/network/connmanager"
"github.com/karlsend/PYVERT/testfork/karlsend/infrastructure/network/netadapter"
)
// Manager manages the p2p protocol
type Manager struct {
context *flowcontext.FlowContext
routersWaitGroup sync.WaitGroup
isClosed uint32
}
// NewManager creates a new instance of the p2p protocol manager
func NewManager(cfg *config.Config, domain domain.Domain, netAdapter *netadapter.NetAdapter, addressManager *addressmanager.AddressManager,
connectionManager *connmanager.ConnectionManager) (*Manager, error) {
manager := Manager{
context: flowcontext.New(cfg, domain, addressManager, netAdapter, connectionManager),
}
netAdapter.SetP2PRouterInitializer(manager.routerInitializer)
return &manager, nil
}
// Close closes the protocol manager and waits until all p2p flows
// finish.
func (m *Manager) Close() {
if !atomic.CompareAndSwapUint32(&m.isClosed, 0, 1) {
panic(errors.New("The protocol manager was already closed"))
}
atomic.StoreUint32(&m.isClosed, 1)
m.context.Close()
m.routersWaitGroup.Wait()
}
// Peers returns the currently active peers
func (m *Manager) Peers() []*peerpkg.Peer {
return m.context.Peers()
}
// IBDPeer returns the current IBD peer or null if the node is not
// in IBD
func (m *Manager) IBDPeer() *peerpkg.Peer {
return m.context.IBDPeer()
}
// AddTransaction adds transaction to the mempool and propagates it.
func (m *Manager) AddTransaction(tx *externalapi.DomainTransaction, allowOrphan bool) error {
return m.context.AddTransaction(tx, allowOrphan)
}
// AddBlock adds the given block to the DAG and propagates it.
func (m *Manager) AddBlock(block *externalapi.DomainBlock) error {
return m.context.AddBlock(block)
}
// Context returns the manager's flow context
func (m *Manager) Context() *flowcontext.FlowContext {
return m.context
}
func (m *Manager) runFlows(flows []*common.Flow, peer *peerpkg.Peer, errChan <-chan error, flowsWaitGroup *sync.WaitGroup) error {
flowsWaitGroup.Add(len(flows))
for _, flow := range flows {
executeFunc := flow.ExecuteFunc // extract to new variable so that it's not overwritten
spawn(fmt.Sprintf("flow-%s", flow.Name), func() {
executeFunc(peer)
flowsWaitGroup.Done()
})
}
return <-errChan
}
// SetOnNewBlockTemplateHandler sets the onNewBlockTemplate handler
func (m *Manager) SetOnNewBlockTemplateHandler(onNewBlockTemplateHandler flowcontext.OnNewBlockTemplateHandler) {
m.context.SetOnNewBlockTemplateHandler(onNewBlockTemplateHandler)
}
// SetOnPruningPointUTXOSetOverrideHandler sets the OnPruningPointUTXOSetOverride handler
func (m *Manager) SetOnPruningPointUTXOSetOverrideHandler(onPruningPointUTXOSetOverrideHandler flowcontext.OnPruningPointUTXOSetOverrideHandler) {
m.context.SetOnPruningPointUTXOSetOverrideHandler(onPruningPointUTXOSetOverrideHandler)
}
// SetOnTransactionAddedToMempoolHandler sets the onTransactionAddedToMempool handler
func (m *Manager) SetOnTransactionAddedToMempoolHandler(onTransactionAddedToMempoolHandler flowcontext.OnTransactionAddedToMempoolHandler) {
m.context.SetOnTransactionAddedToMempoolHandler(onTransactionAddedToMempoolHandler)
}
// IsIBDRunning returns true if IBD is currently marked as running
func (m *Manager) IsIBDRunning() bool {
return m.context.IsIBDRunning()
}