forked from hyperledger/fabric
-
Notifications
You must be signed in to change notification settings - Fork 0
/
deliveryclient.go
185 lines (163 loc) · 6.25 KB
/
deliveryclient.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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
/*
Copyright IBM Corp. All Rights Reserved.
SPDX-License-Identifier: Apache-2.0
*/
package deliverservice
import (
"context"
"crypto/x509"
"errors"
"fmt"
"sync"
"time"
"github.com/hyperledger/fabric-protos-go/orderer"
"github.com/hyperledger/fabric/common/flogging"
"github.com/hyperledger/fabric/common/util"
"github.com/hyperledger/fabric/core/comm"
"github.com/hyperledger/fabric/internal/pkg/identity"
"github.com/hyperledger/fabric/internal/pkg/peer/blocksprovider"
"github.com/hyperledger/fabric/internal/pkg/peer/orderers"
"google.golang.org/grpc"
)
var logger = flogging.MustGetLogger("deliveryClient")
// DeliverService used to communicate with orderers to obtain
// new blocks and send them to the committer service
type DeliverService interface {
// StartDeliverForChannel dynamically starts delivery of new blocks from ordering service
// to channel peers.
// When the delivery finishes, the finalizer func is called
StartDeliverForChannel(chainID string, ledgerInfo blocksprovider.LedgerInfo, finalizer func()) error
// StopDeliverForChannel dynamically stops delivery of new blocks from ordering service
// to channel peers.
StopDeliverForChannel(chainID string) error
// Stop terminates delivery service and closes the connection
Stop()
}
// deliverServiceImpl the implementation of the delivery service
// maintains connection to the ordering service and maps of
// blocks providers
type deliverServiceImpl struct {
conf *Config
blockProviders map[string]*blocksprovider.Deliverer
lock sync.RWMutex
stopping bool
}
// Config dictates the DeliveryService's properties,
// namely how it connects to an ordering service endpoint,
// how it verifies messages received from it,
// and how it disseminates the messages to other peers
type Config struct {
IsStaticLeader bool
// CryptoSvc performs cryptographic actions like message verification and signing
// and identity validation.
CryptoSvc blocksprovider.BlockVerifier
// Gossip enables to enumerate peers in the channel, send a message to peers,
// and add a block to the gossip state transfer layer.
Gossip blocksprovider.GossipServiceAdapter
// OrdererSource provides orderer endpoints, complete with TLS cert pools.
OrdererSource *orderers.ConnectionSource
// Signer is the identity used to sign requests.
Signer identity.SignerSerializer
// GRPC Client
DeliverGRPCClient *comm.GRPCClient
// Configuration values for deliver service.
// TODO: merge 2 Config struct
DeliverServiceConfig *DeliverServiceConfig
}
// NewDeliverService construction function to create and initialize
// delivery service instance. It tries to establish connection to
// the specified in the configuration ordering service, in case it
// fails to dial to it, return nil
func NewDeliverService(conf *Config) DeliverService {
ds := &deliverServiceImpl{
conf: conf,
blockProviders: make(map[string]*blocksprovider.Deliverer),
}
return ds
}
type DialerAdapter struct {
Client *comm.GRPCClient
}
func (da DialerAdapter) Dial(address string, certPool *x509.CertPool) (*grpc.ClientConn, error) {
return da.Client.NewConnection(address, comm.CertPoolOverride(certPool))
}
type DeliverAdapter struct{}
func (DeliverAdapter) Deliver(ctx context.Context, clientConn *grpc.ClientConn) (orderer.AtomicBroadcast_DeliverClient, error) {
return orderer.NewAtomicBroadcastClient(clientConn).Deliver(ctx)
}
// StartDeliverForChannel starts blocks delivery for channel
// initializes the grpc stream for given chainID, creates blocks provider instance
// that spawns in go routine to read new blocks starting from the position provided by ledger
// info instance.
func (d *deliverServiceImpl) StartDeliverForChannel(chainID string, ledgerInfo blocksprovider.LedgerInfo, finalizer func()) error {
d.lock.Lock()
defer d.lock.Unlock()
if d.stopping {
errMsg := fmt.Sprintf("Delivery service is stopping cannot join a new channel %s", chainID)
logger.Errorf(errMsg)
return errors.New(errMsg)
}
if _, exist := d.blockProviders[chainID]; exist {
errMsg := fmt.Sprintf("Delivery service - block provider already exists for %s found, can't start delivery", chainID)
logger.Errorf(errMsg)
return errors.New(errMsg)
}
logger.Info("This peer will retrieve blocks from ordering service and disseminate to other peers in the organization for channel", chainID)
dc := &blocksprovider.Deliverer{
ChannelID: chainID,
Gossip: d.conf.Gossip,
Ledger: ledgerInfo,
BlockVerifier: d.conf.CryptoSvc,
Dialer: DialerAdapter{
Client: d.conf.DeliverGRPCClient,
},
Orderers: d.conf.OrdererSource,
DoneC: make(chan struct{}),
Signer: d.conf.Signer,
DeliverStreamer: DeliverAdapter{},
Logger: flogging.MustGetLogger("peer.blocksprovider").With("channel", chainID),
MaxRetryDelay: time.Duration(d.conf.DeliverServiceConfig.ReConnectBackoffThreshold),
MaxRetryDuration: d.conf.DeliverServiceConfig.ReconnectTotalTimeThreshold,
InitialRetryDelay: 100 * time.Millisecond,
YieldLeadership: !d.conf.IsStaticLeader,
}
if d.conf.DeliverGRPCClient.MutualTLSRequired() {
dc.TLSCertHash = util.ComputeSHA256(d.conf.DeliverGRPCClient.Certificate().Certificate[0])
}
d.blockProviders[chainID] = dc
go func() {
dc.DeliverBlocks()
finalizer()
}()
return nil
}
// StopDeliverForChannel stops blocks delivery for channel by stopping channel block provider
func (d *deliverServiceImpl) StopDeliverForChannel(chainID string) error {
d.lock.Lock()
defer d.lock.Unlock()
if d.stopping {
errMsg := fmt.Sprintf("Delivery service is stopping, cannot stop delivery for channel %s", chainID)
logger.Errorf(errMsg)
return errors.New(errMsg)
}
client, exist := d.blockProviders[chainID]
if !exist {
errMsg := fmt.Sprintf("Delivery service - no block provider for %s found, can't stop delivery", chainID)
logger.Errorf(errMsg)
return errors.New(errMsg)
}
client.Stop()
delete(d.blockProviders, chainID)
logger.Debug("This peer will stop pass blocks from orderer service to other peers")
return nil
}
// Stop all service and release resources
func (d *deliverServiceImpl) Stop() {
d.lock.Lock()
defer d.lock.Unlock()
// Marking flag to indicate the shutdown of the delivery service
d.stopping = true
for _, client := range d.blockProviders {
client.Stop()
}
}