|
| 1 | +/* |
| 2 | +Copyright IBM Corp. All Rights Reserved. |
| 3 | +
|
| 4 | +SPDX-License-Identifier: Apache-2.0 |
| 5 | +*/ |
| 6 | + |
| 7 | +package main |
| 8 | + |
| 9 | +import ( |
| 10 | + "context" |
| 11 | + "flag" |
| 12 | + "fmt" |
| 13 | + "io/ioutil" |
| 14 | + "math" |
| 15 | + "os" |
| 16 | + "strings" |
| 17 | + "time" |
| 18 | + |
| 19 | + "github.com/hyperledger/fabric/common/crypto" |
| 20 | + "github.com/hyperledger/fabric/common/flogging" |
| 21 | + "github.com/hyperledger/fabric/common/localmsp" |
| 22 | + genesisconfig "github.com/hyperledger/fabric/common/tools/configtxgen/localconfig" |
| 23 | + "github.com/hyperledger/fabric/common/tools/protolator" |
| 24 | + "github.com/hyperledger/fabric/common/util" |
| 25 | + "github.com/hyperledger/fabric/core/comm" |
| 26 | + config2 "github.com/hyperledger/fabric/core/config" |
| 27 | + "github.com/hyperledger/fabric/msp" |
| 28 | + common2 "github.com/hyperledger/fabric/peer/common" |
| 29 | + "github.com/hyperledger/fabric/protos/common" |
| 30 | + "github.com/hyperledger/fabric/protos/orderer" |
| 31 | + "github.com/hyperledger/fabric/protos/peer" |
| 32 | + "github.com/hyperledger/fabric/protos/utils" |
| 33 | + "github.com/spf13/viper" |
| 34 | +) |
| 35 | + |
| 36 | +var ( |
| 37 | + channelID string |
| 38 | + serverAddr string |
| 39 | + clientKeyPath string |
| 40 | + clientCertPath string |
| 41 | + serverRootCAPath string |
| 42 | + seek int |
| 43 | + quiet bool |
| 44 | + filtered bool |
| 45 | + tlsEnabled bool |
| 46 | + mTlsEnabled bool |
| 47 | + |
| 48 | + oldest = &orderer.SeekPosition{Type: &orderer.SeekPosition_Oldest{Oldest: &orderer.SeekOldest{}}} |
| 49 | + newest = &orderer.SeekPosition{Type: &orderer.SeekPosition_Newest{Newest: &orderer.SeekNewest{}}} |
| 50 | + maxStop = &orderer.SeekPosition{Type: &orderer.SeekPosition_Specified{Specified: &orderer.SeekSpecified{Number: math.MaxUint64}}} |
| 51 | +) |
| 52 | + |
| 53 | +const ( |
| 54 | + OLDEST = -2 |
| 55 | + NEWEST = -1 |
| 56 | + |
| 57 | + ROOT = "core" |
| 58 | +) |
| 59 | + |
| 60 | +var logger = flogging.MustGetLogger("eventsclient") |
| 61 | + |
| 62 | +// deliverClient abstracts common interface |
| 63 | +// for deliver and deliverfiltered services |
| 64 | +type deliverClient interface { |
| 65 | + Send(*common.Envelope) error |
| 66 | + Recv() (*peer.DeliverResponse, error) |
| 67 | +} |
| 68 | + |
| 69 | +// eventsClient client to get connected to the |
| 70 | +// events peer delivery system |
| 71 | +type eventsClient struct { |
| 72 | + client deliverClient |
| 73 | + signer crypto.LocalSigner |
| 74 | + tlsCertHash []byte |
| 75 | +} |
| 76 | + |
| 77 | +func (r *eventsClient) seekOldest() error { |
| 78 | + return r.client.Send(r.seekHelper(oldest, maxStop)) |
| 79 | +} |
| 80 | + |
| 81 | +func (r *eventsClient) seekNewest() error { |
| 82 | + return r.client.Send(r.seekHelper(newest, maxStop)) |
| 83 | +} |
| 84 | + |
| 85 | +func (r *eventsClient) seekSingle(blockNumber uint64) error { |
| 86 | + specific := &orderer.SeekPosition{Type: &orderer.SeekPosition_Specified{Specified: &orderer.SeekSpecified{Number: blockNumber}}} |
| 87 | + return r.client.Send(r.seekHelper(specific, specific)) |
| 88 | +} |
| 89 | + |
| 90 | +func (r *eventsClient) seekHelper(start *orderer.SeekPosition, stop *orderer.SeekPosition) *common.Envelope { |
| 91 | + env, err := utils.CreateSignedEnvelopeWithTLSBinding(common.HeaderType_DELIVER_SEEK_INFO, channelID, r.signer, &orderer.SeekInfo{ |
| 92 | + Start: start, |
| 93 | + Stop: stop, |
| 94 | + Behavior: orderer.SeekInfo_BLOCK_UNTIL_READY, |
| 95 | + }, 0, 0, r.tlsCertHash) |
| 96 | + if err != nil { |
| 97 | + panic(err) |
| 98 | + } |
| 99 | + return env |
| 100 | +} |
| 101 | + |
| 102 | +func (r *eventsClient) readEventsStream() { |
| 103 | + for { |
| 104 | + msg, err := r.client.Recv() |
| 105 | + if err != nil { |
| 106 | + logger.Info("Error receiving:", err) |
| 107 | + return |
| 108 | + } |
| 109 | + |
| 110 | + switch t := msg.Type.(type) { |
| 111 | + case *peer.DeliverResponse_Status: |
| 112 | + logger.Info("Got status ", t) |
| 113 | + return |
| 114 | + case *peer.DeliverResponse_Block: |
| 115 | + if !quiet { |
| 116 | + logger.Info("Received block: ") |
| 117 | + err := protolator.DeepMarshalJSON(os.Stdout, t.Block) |
| 118 | + if err != nil { |
| 119 | + fmt.Printf(" Error pretty printing block: %s", err) |
| 120 | + } |
| 121 | + } else { |
| 122 | + logger.Info("Received block: ", t.Block.Header.Number) |
| 123 | + } |
| 124 | + case *peer.DeliverResponse_FilteredBlock: |
| 125 | + if !quiet { |
| 126 | + logger.Info("Received filtered block: ") |
| 127 | + err := protolator.DeepMarshalJSON(os.Stdout, t.FilteredBlock) |
| 128 | + if err != nil { |
| 129 | + fmt.Printf(" Error pretty printing filtered block: %s", err) |
| 130 | + } |
| 131 | + } else { |
| 132 | + logger.Info("Received filtered block: ", t.FilteredBlock.Number) |
| 133 | + } |
| 134 | + } |
| 135 | + } |
| 136 | +} |
| 137 | + |
| 138 | +func (r *eventsClient) seek(s int) error { |
| 139 | + var err error |
| 140 | + switch seek { |
| 141 | + case OLDEST: |
| 142 | + err = r.seekOldest() |
| 143 | + case NEWEST: |
| 144 | + err = r.seekNewest() |
| 145 | + default: |
| 146 | + err = r.seekSingle(uint64(seek)) |
| 147 | + } |
| 148 | + return err |
| 149 | +} |
| 150 | + |
| 151 | +func main() { |
| 152 | + initConfig() |
| 153 | + initMSP() |
| 154 | + readCLInputs() |
| 155 | + |
| 156 | + if seek < OLDEST { |
| 157 | + logger.Info("Invalid seek value") |
| 158 | + flag.PrintDefaults() |
| 159 | + return |
| 160 | + } |
| 161 | + |
| 162 | + clientConfig := comm.ClientConfig{ |
| 163 | + KaOpts: comm.DefaultKeepaliveOptions(), |
| 164 | + SecOpts: &comm.SecureOptions{}, |
| 165 | + Timeout: 5 * time.Minute, |
| 166 | + } |
| 167 | + |
| 168 | + if tlsEnabled { |
| 169 | + clientConfig.SecOpts.UseTLS = true |
| 170 | + rootCert, err := ioutil.ReadFile(serverRootCAPath) |
| 171 | + if err != nil { |
| 172 | + logger.Info("error loading TLS root certificate", err) |
| 173 | + return |
| 174 | + } |
| 175 | + clientConfig.SecOpts.ServerRootCAs = [][]byte{rootCert} |
| 176 | + if mTlsEnabled { |
| 177 | + clientConfig.SecOpts.RequireClientCert = true |
| 178 | + clientKey, err := ioutil.ReadFile(clientKeyPath) |
| 179 | + if err != nil { |
| 180 | + logger.Info("error loading client TLS key from", clientKeyPath) |
| 181 | + return |
| 182 | + } |
| 183 | + clientConfig.SecOpts.Key = clientKey |
| 184 | + |
| 185 | + clientCert, err := ioutil.ReadFile(clientCertPath) |
| 186 | + if err != nil { |
| 187 | + logger.Info("error loading client TLS certificate from path", clientCertPath) |
| 188 | + return |
| 189 | + } |
| 190 | + clientConfig.SecOpts.Certificate = clientCert |
| 191 | + } |
| 192 | + } |
| 193 | + |
| 194 | + grpcClient, err := comm.NewGRPCClient(clientConfig) |
| 195 | + if err != nil { |
| 196 | + logger.Info("Error creating grpc client:", err) |
| 197 | + return |
| 198 | + } |
| 199 | + conn, err := grpcClient.NewConnection(serverAddr, "") |
| 200 | + if err != nil { |
| 201 | + logger.Info("Error connecting:", err) |
| 202 | + return |
| 203 | + } |
| 204 | + |
| 205 | + var client deliverClient |
| 206 | + if filtered { |
| 207 | + client, err = peer.NewDeliverClient(conn).DeliverFiltered(context.Background()) |
| 208 | + } else { |
| 209 | + client, err = peer.NewDeliverClient(conn).Deliver(context.Background()) |
| 210 | + } |
| 211 | + |
| 212 | + if err != nil { |
| 213 | + logger.Info("Error connecting:", err) |
| 214 | + return |
| 215 | + } |
| 216 | + |
| 217 | + events := &eventsClient{ |
| 218 | + client: client, |
| 219 | + signer: localmsp.NewSigner(), |
| 220 | + } |
| 221 | + |
| 222 | + if mTlsEnabled { |
| 223 | + events.tlsCertHash = util.ComputeSHA256(grpcClient.Certificate().Certificate[0]) |
| 224 | + } |
| 225 | + |
| 226 | + events.seek(seek) |
| 227 | + if err != nil { |
| 228 | + logger.Info("Received error:", err) |
| 229 | + return |
| 230 | + } |
| 231 | + |
| 232 | + events.readEventsStream() |
| 233 | +} |
| 234 | + |
| 235 | +func readCLInputs() { |
| 236 | + flag.StringVar(&serverAddr, "server", "localhost:7051", "The RPC server to connect to.") |
| 237 | + flag.StringVar(&channelID, "channelID", genesisconfig.TestChainID, "The channel ID to deliver from.") |
| 238 | + flag.BoolVar(&quiet, "quiet", false, "Only print the block number, will not attempt to print its block contents.") |
| 239 | + flag.BoolVar(&filtered, "filtered", true, "Whenever to read filtered events from the peer delivery service or get regular blocks.") |
| 240 | + flag.BoolVar(&tlsEnabled, "tls", false, "TLS enabled/disabled") |
| 241 | + flag.BoolVar(&mTlsEnabled, "mTls", false, "Mutual TLS enabled/disabled (whenever server side validates clients TLS certificate)") |
| 242 | + flag.StringVar(&clientKeyPath, "clientKey", "", "Specify path to the client TLS key") |
| 243 | + flag.StringVar(&clientCertPath, "clientCert", "", "Specify path to the client TLS certificate") |
| 244 | + flag.StringVar(&serverRootCAPath, "rootCert", "", "Specify path to the server root CA certificate") |
| 245 | + flag.IntVar(&seek, "seek", OLDEST, "Specify the range of requested blocks."+ |
| 246 | + "Acceptable values:"+ |
| 247 | + "-2 (or -1) to start from oldest (or newest) and keep at it indefinitely."+ |
| 248 | + "N >= 0 to fetch block N only.") |
| 249 | + flag.Parse() |
| 250 | +} |
| 251 | + |
| 252 | +func initMSP() { |
| 253 | + // Init the MSP |
| 254 | + var mspMgrConfigDir = config2.GetPath("peer.mspConfigPath") |
| 255 | + var mspID = viper.GetString("peer.localMspId") |
| 256 | + var mspType = viper.GetString("peer.localMspType") |
| 257 | + if mspType == "" { |
| 258 | + mspType = msp.ProviderTypeToString(msp.FABRIC) |
| 259 | + } |
| 260 | + err := common2.InitCrypto(mspMgrConfigDir, mspID, mspType) |
| 261 | + if err != nil { // Handle errors reading the config file |
| 262 | + panic(fmt.Sprintf("Cannot run client because %s", err.Error())) |
| 263 | + } |
| 264 | +} |
| 265 | + |
| 266 | +func initConfig() { |
| 267 | + // For environment variables. |
| 268 | + viper.SetEnvPrefix(ROOT) |
| 269 | + viper.AutomaticEnv() |
| 270 | + replacer := strings.NewReplacer(".", "_") |
| 271 | + viper.SetEnvKeyReplacer(replacer) |
| 272 | + |
| 273 | + err := common2.InitConfig(ROOT) |
| 274 | + if err != nil { // Handle errors reading the config file |
| 275 | + panic(fmt.Errorf("fatal error when initializing %s config : %s", ROOT, err)) |
| 276 | + } |
| 277 | +} |
0 commit comments