Hyperledger Fabric——balance transfer（三）创建和加入Channel

创建Channel

1.首先看app.js的路由函数

var createChannel = require('./app/create-channel.js');

app.post('/channels', async function(req, res) {

    // 接收参数channel名称和配置文件的路径

    // 通道配置交易路径： ./artifacts/channel/mychannel.tx

    var channelName = req.body.channelName;

    var channelConfigPath = req.body.channelConfigPath;

    if (!channelName) {

        res.json(getErrorMessage('\'channelName\''));

        return;

    }

    if (!channelConfigPath) {

        res.json(getErrorMessage('\'channelConfigPath\''));

        return;

    }

    // 创建通道(username和orgname参数从token中获取)

    let message = await createChannel.createChannel(channelName, channelConfichannelNamegPath, req.username, req.orgname);

    res.send(message);

});

2.再来看create-channel.js中的createChannel()方法

// 通过发送交易的方法向orderer节点发送创建channel的请求

var createChannel = async function(channelName, channelConfigPath, username, orgName) {

    try {

        // 生成组织的客户端实例

        var client = await helper.getClientForOrg(orgName);

        // 从channel配置交易（mychannel.tx）读取源字节

        var envelope = fs.readFileSync(path.join(__dirname, channelConfigPath));

        // 提取channel配置信息

        var channelConfig = client.extractChannelConfig(envelope);

        // 客户端使用 组织管理员 的私钥 签名，作为背书（endorsement）

        // "../artifacts/channel/crypto-config/peerOrganizations/org1.example.com/users/[email protected]/msp/signcerts"

        // 在内部通过_getSigningIdentity(true)获取org admin的私钥 进行签名

        let signature = client.signChannelConfig(channelConfig);

        // 请求结构：

        let request = {

            config: channelConfig,

            signatures: [signature],

            name: channelName,

            txId: client.newTransactionID(true)

        };

        // 发送给orderer节点

        var response = await client.createChannel(request)

        if (response && response.status === 'SUCCESS') {

            logger.debug('Successfully created the channel.');

            let response = {

                success: true,

                message: 'Channel \'' + channelName + '\' created Successfully'

            };

            return response;

        } else {

            throw new Error('Failed to create the channel \'' + channelName + '\'');

        }

    } catch (err) {

        logger.error('Failed to initialize the channel: ' + err.stack ? err.stack : err);

        throw new Error('Failed to initialize the channel: ' + err.toString());

    }

};

加入节点到Channel

1.首先看app.js里的路由函数

// 首先看app.js中的路由函数

app.post('/channels/:channelName/peers', async function(req, res) {

    // 接收参数 channel名称（URL中）和peers（body中）

    var channelName = req.params.channelName;

    var peers = req.body.peers;

    if (!channelName) {

        res.json(getErrorMessage('\'channelName\''));

        return;

    }

    if (!peers || peers.length == 0) {

        res.json(getErrorMessage('\'peers\''));

        return;

    }

    // 将peers加入channel

    let message =  await join.joinChannel(channelName, peers, req.username, req.orgname);

    res.send(message);

});

2.然后看join-channel.js中的joinChannel()函数

// 然后看join-channel.js中的joinChannel()函数

var joinChannel = async function(channel_name, peers, username, org_name) {

    var error_message = null;

    var all_eventhubs = [];

    try {

        // 首先创建client对象

        var client = await helper.getClientForOrg(org_name, username);

        //  生成channel对象

        var channel = client.getChannel(channel_name);

        if(!channel) {

            let message = util.format('Channel %s was not defined in the connection profile', channel_name);

            logger.error(message);

            throw new Error(message);

        }

        // 生成基于组织管理员的 TransactionID 对象

        // 该对象包含一个随机数（nonce）以及t(nonce + signingIdentity)的hash值

        let request = {

            txId :  client.newTransactionID(true)

        };

        // 从orderer节点获取创世区块（genesis block）

        let genesis_block = await channel.getGenesisBlock(request);

        // 告诉节点加入channel，等待每个节点的event hub通知我们节点加入channel成功

        var promises = [];

        var block_registration_numbers = [];

        // 给组织中的每个节点分配一个EventHub对象

        let event_hubs = client.getEventHubsForOrg(org_name);

        #event_hubs.forEach((eh) => {

            #let configBlockPromise = new Promise((resolve, reject) => {

                // 设置超时时间

                #let event_timeout = setTimeout(() => {

                    let message = 'REQUEST_TIMEOUT:' + eh._ep._endpoint.addr;

                    logger.error(message);

                    eh.disconnect();

                    reject(new Error(message));

                }, 60000);

                // 注册区块监听

                #let block_registration_number = eh.registerBlockEvent((block) => {

                    clearTimeout(event_timeout);

                    // 配置区块中只能有一个交易

                    if (block.data.data.length === 1) {

                        // 一个节点可能属于多个channel

                        // 所以需要检验该区块是否来自目标channel

                        var channel_header = block.data.data[0].payload.header.channel_header;

                        if (channel_header.channel_id === channel_name) {

                            let message = util.format('EventHub %s has reported a block update for channel %s',eh._ep._endpoint.addr,channel_name);

                            logger.info(message)

                            resolve(message);

                        } else {

                            let message = util.format('Unknown channel block event received from %s',eh._ep._endpoint.addr);

                            logger.error(message);

                            reject(new Error(message));

                        }

                    }

                }, (err) => {

                    clearTimeout(event_timeout);

                    let message = 'Problem setting up the event hub :'+ err.toString();

                    logger.error(message);

                    reject(new Error(message));

                });

                // 每一个client实例对应一个注册number，后面将作为参数对监听进行注销

                block_registration_numbers.push(block_registration_number);

                // 保存EventHub对象，方便后面对事件流断开连接

                all_eventhubs.push(eh);

            });

            promises.push(configBlockPromise);

            // 开启事件流

            eh.connect();

        });

        // 封装加入channel的请求

        let join_request = {

            targets: peers,                             // 要加入channel的节点

            txId: client.newTransactionID(true),        // 基于组织管理员的TransactionID

            block: genesis_block                        // 创世区块

        };

        // 调用SDK中的joinChannel()方法，主要是通过sendProposal()将

        // 加入channel的交易提案发送给背书节点进行背书

        let join_promise = channel.joinChannel(join_request);

        // 保存返回结果：提案响应（ProposalResponse）的Promise

        promises.push(join_promise);

        let results = await Promise.all(promises);

        logger.debug(util.format('Join Channel RESPONSE : %j', results));

        // 检查所有Promise返回（包括监听事件和发送join请求）

        // 只要有一个结果异常则宣布join channel失败

        let peers_results = results.pop();

        for(let i in peers_results) {

            let peer_result = peers_results[i];

            if(peer_result.response && peer_result.response.status == 200) {

                logger.info('Successfully joined peer to the channel %s',channel_name);

            } else {

                let message = util.format('Failed to joined peer to the channel %s',channel_name);

                error_message = message;

                logger.error(message);

            }

        }

        // 查看事件中心的消息报告

        for(let i in results) {

            let event_hub_result = results[i];

            let event_hub = event_hubs[i];

            let block_registration_number = block_registration_numbers[i];

            logger.debug('Event results for event hub :%s',event_hub._ep._endpoint.addr);

            if(typeof event_hub_result === 'string') {

                logger.debug(event_hub_result);

            } else {

                if(!error_message) error_message = event_hub_result.toString();

                logger.debug(event_hub_result.toString());

            }

            // 注销事件监听

            event_hub.unregisterBlockEvent(block_registration_number);

        }

    } catch(error) {

        logger.error('Failed to join channel due to error: ' + error.stack ? error.stack : error);

        error_message = error.toString();

    }

    // 断开所有event hub事件流

    all_eventhubs.forEach((eh) => {

        eh.disconnect();

    });

};

测试

创建channel

curl -s -X POST \

  http://localhost:4000/channels \

  -H "authorization: Bearer $ORG1_TOKEN" \

  -H "content-type: application/json" \

  -d '{

    "channelName":"mychannel",

    "channelConfigPath":"../artifacts/channel/mychannel.tx"

}'

返回结果

{"success":true,"message":"Channel 'mychannel' created Successfully"}

将Org1中的两个peer加入channel

curl -s -X POST \

  http://localhost:4000/channels/mychannel/peers \

  -H "authorization: Bearer $ORG1_TOKEN" \

  -H "content-type: application/json" \

  -d '{

    "peers": ["peer0.org1.example.com","peer1.org1.example.com"]

}'

返回结果

{"success":true,"message":"Successfully joined peers in organization Org1 to the channel:mychannel"}

总结

创建channel：

生成client对象，从channel配置交易mychannel.tx中提取配置信息
利用组织admin的私钥进行签名，这个过程相当于背书
封装请求结构（包括channel配置信息，对配置信息的签名，channelName，基于admin的txId), 并调用 clinet.createChannel()将请求发送到orderer节点
实际上在内部调用orderer.sendBroadcast()，向order service广播交易，orderer将配置交易打包成区块后广播给各记账节点，记入账本中。（这个配置区块就是创建的channel所对应的chain的创世区块）

加入channel：

生成client对象，生成channel对象
从orderer节点获取genesis block （调用 orderer.sendDeliver() 接口）
对要加入channel的每个peer设置事件区块监听，根据指定channel是否生成了新的配置区块来判断join channel过程是否成功
封装请求结构（包括目标peers，基于组织admin的txId，genesis block），调用channel.joinChannel()方法，内部是通过peer.sendProposal()将该交易提案发送给背书节点进行背书，返回背书响应Promise

客户端将背书过的交易发送到orderer服务，排序后广播给记账节点，将该配置区块记录到ledger中，此时EventHub监听到该事件。
加入channel成功后，注销区块事件监听，断开所有event hub事件流。