FISCO-BCOS Web3SDK 1.3与2.0对比

2020-04-30 本文已影响0人 CodingCattwo

原创内容，未经允许不得转载

本文对源码的解读为个人理解，欢迎指正
关联github仓库：FISCO-BCOS/web3sdk

FISCO BCOS 1.3的sdk(proxy)

web3sdk 1.3

client （打断点执行一遍初始化过程)

p12 pem manager 私钥管理
ResponseCallback
- onResponse
- onTimeout
- retrySendMessage 遍历客户端节点，尝试重发
channelPushCallback
TransactionSucCallback

Service 客户端的Service

setTopic flushTopics(遍历channelConnections)
run 调用channelConnections init()等方法 -> ThreadPoolTaskExecutor ( for循环) getNetworkConnection保持长连接
包含sendEthMessage获取Callback sendResponseMessage2 onReceiveEthereumMessage(ConnectionCallback)

// 初始化GroupChannelConnectionsConfig channel配置
    List<ChannelConnections> channelConnectionsList = new ArrayList<>();
    List<String> connectionsList = new ArrayList<>();
    connectionsList.add(ip + ":" + channelPort);
    ChannelConnections channelConnections = new ChannelConnections();
    channelConnections.setConnectionsStr(connectionsList);
    channelConnections.setGroupId(1);
    channelConnectionsList.add(channelConnections);

    GroupChannelConnectionsConfig groupChannelConnectionsConfig =
            new GroupChannelConnectionsConfig();
    groupChannelConnectionsConfig.setAllChannelConnections(channelConnectionsList);
// service以Bean groupChannelConnectionsConfig初始化，并调用.run()
    Service service = new Service();
    ...
    service.setAllChannelConnections(groupChannelConnectionsConfig);
    service.run(); // run直到channleConnections连接成功，否则返回连接失败
// 以service初始化Web3j实例
    ChannelEthereumService channelEthereumService = new ChannelEthereumService();
    ...
    channelEthereumService.setChannelService(service);
    Web3j web3j = Web3j.build(channelEthereumService);

Handler：除了channelConnections是发起连接时的，其余为连接后触发

channelConnections: 上文client中用到的connection实例，需要ca。大致调用顺序：

init()
- initDefaultCertConfig
startConnect
initSslContextForListening
startConnect() -> start netty(boostrap) -> initSslContextForConnect -> 初始化后，断连则reconnect

SslContext sslCtx =
    SslContextBuilder.forClient()
            .trustManager(caInputStream)
            .keyManager(
                    keystorecaResource.getInputStream(),
                    keystorekeyResource.getInputStream())
            .sslProvider(SslProvider.JDK)
            .build();

ChannelHandler 监测channel是否还active，userEventTriggered检测event是读或写
ConnectionCallback 连接时返回处理
- onConnect
- queryBlockNumberForSelectNodes
Encoder Decoder工具类
Message ConnectionInfo实体类
GroupChannelConnectionsConfig设置allConnections

proxy (2.0后移除，改为protocol协议+event事件通知)

ConnectionPair 此处为RemoteChannelConnections

public ChannelHandlerContext localConnection; //SDK连接
public ChannelHandlerContext remoteConnection; //节点连接
# ==为何有两个连接：一个与本地交互，一个与远端交互==

init() // 初始化connections 与2.0的channelConnection一样
retrySendRemoteMessage -> 随机往节点sdk发remoteConnectionInfos.get(random) -> 多次失败后往本地刷新错误信息

ByteBuf out = localConnection.alloc().buffer();
message.writeHeader(out);
message.writeExtra(out);
localConnection.writeAndFlush(out);

Server

内部类ConnectionCallBack
- onMessage 处理多种msg
- onConnect 连接到新节点 onDisconnect 有sdk断开
```
# onConnect是与节点rpc交互还是和节点的sdk交互
```
Server类的变量

private ChannelConnections localConnections = new ChannelConnections();
private ChannelConnections remoteConnections;
// 记录来自本地的Request或远端的Push
private Map<String, ConnectionPair> seq2Connections = new ConcurrentHashMap<String, ConnectionPair>();

run() 初始化ProxyServer 分别是本地的ConnectionCallBack和远端的ConnectionCallBack

ConnectionCallback localConnectionCallback = new ConnectionCallback();
localConnectionCallback.setServer(this);
localConnectionCallback.setFromRemote(false);

ConnectionCallback remoteConnectionCallback = new ConnectionCallback();
remoteConnectionCallback.setServer(this);
remoteConnectionCallback.setFromRemote(true);

localConnections.setCallback(localConnectionCallback);
localConnections.startListen(bindPort);

remoteConnections.setCallback(remoteConnectionCallback);
remoteConnections.init();
remoteConnections.setThreadPool(threadPool);
remoteConnections.startConnect();

onTopic(ChannelHandlerContext ctx, Message message) 根据ChannelHandlerContext获得指定localConnection, 设置localConnection.ConnectionInfo的topics后广播

info.setTopics(topics);
broadcastTopic();

broadcastTopic (netty的ChannelHandlerContext.writeAndFlush 广播到所有远端节点)
- 先从localConnections遍历ConnectionInfos，将有效的连接添加到全局topic
- 广播给所有远端或发送指定的远端 remoteConnections

// 本地查找connections获得pair，用于响应onLocal和onRemote
ConnectionPair pair = seq2Connections.get(message.getSeq());

onLocalMessage 响应本地消息，使用remoteConnection发往远端，判断seq2Connections是否已包含该ConnectionPair

如果已存在pair，则是localMessage要发回应给远端

// 发送到远端的响应
remoteCtx = pair.remoteConnection;

如果不存在pair，则可能是本地新发的请求或者收到远端的push

pair = new ConnectionPair();
pair.localConnection = ctx;
pair.setServer(this);
pair.setMessage(message);

...
// 根据nodeId拿到remoteConnection
remoteCtx = remoteConnections.getNetworkConnectionByHost(conn.getHost(), conn.getPort());
pair.remoteConnection = remoteCtx;

onRemoteMessage 响应远端消息，使用localConnection发往本地

if(message.getType() == 0x30)该消息类型则从topic找已有的ctx赋予connection

// pair存在 需要找到关注该topic的连接
ChannelHandlerContext topicCtx = localConnections.getNetworkConnectionByHost
.. 
// pair不存在，随机下发
Random random = new SecureRandom();
Integer index = random.nextInt(topicCtxs.size());
pair.localConnection =  (ChannelHandlerContext) topicCtxs.toArray()[index];;
pair.remoteConnection = ctx;

非0x30的message

pair已存在：收到远端回包消息（本地向远端发送响应或者）

// 收到来自远端的回包
localCtx = pair.localConnection;
pair.retrySendRemoteMessage();

不存在Pair，向远端新发请求或者新收到远端的push

//其他消息（链上链下一期），随机发
localCtx = localConnections.randomNetworkConnection();

onHeartBeat

dto

ChannelMessage实体类，ChannelPush实体类
ChannelRequest/Response实体类
BcosMessage实体类
BcosRequest/Response实体类

FISCO BCOS 2.0的web3sdk(channel)

底层配合也增加了libchannelserver 的模块，包含channelServer，sdk则作为channleClient

web3sdk 2.0

client

同之前的CallBack和p12 pem manager 私钥管理
- XXCallback
  - onResponse
  - onTimeout
  - retrySendMessage 遍历客户端节点，尝试重发
- channelPushCallback
- TransactionSucCallback
新增了BlockNotifyCallBack接口
- onBlockNotify(int groupID, BigInteger blockNumber)
AMOPVerifyUtil, ECDSAUtil,ReceiptEncoder等工具
新增交易相关
- ExecuteTransaction 继承Contract类
  - executeTransaction(Function)
  - asyncExecuteTransaction(function, callback)
- CallContract, Callresult call调用合约返回Callresult，也包含execTx
Merkle
- calculateMerkleRoot

TransactionResource

getTransactionWithProof调用web3j获取txWithProof，本地计算merkle

String proof =
    Merkle.calculateMerkleRoot(
            transactionReceiptWithProof
                    .getTransactionReceiptWithProof()
                    .getReceiptProof(),
            input);

Service
- 根据AMOP等加入了process方法
  - signForAmop, checkSignForAmop
- 去除了原有sdk server和节点的代码逻辑，加入了protocol事件通知协议的代码：sendCheckResultToNode
- 增加事件通知EventLogFilter的方法: registerEventLogFilter可以注册事件通知
  - register的时候asyncSendRegisterEventLogFilterMessage(EventLogFilter filter)往链上发送注册消息，处理多种注册结果
- 处理事件的Push和response，本地往链上发送则push，链上往本地推送则接受response
- callback,分别是AMOP, channel协议消息，event消息，出块消息，tx消息等(channel协议包含了原生的EthereumMessage也就是BcosMessage，以及自定义的ChannelMessage2)
  - onReceiveChannelMessage2 (ChannelResponseCallback2) AMOP
  - onReceiveEthereumMessage (BcosResponseCallback)
  - onReceiveBlockNotify(ChannelHandlerContext ctx, ChannelMessage2 message) 可以重写，并通过setBlockNotifyCallBack方法来获取出块推送
  - onReceiveHeartbeat
  - onReceiveTransactionMessage
```
EventLogPushCallback callback =
                (EventLogPushCallback)
                        eventLogFilterManager.getFilterCallback(resp.getFilterID());
 --> callback.onPushEventLog(
                        EventLogFilterPushStatus.SUCCESS.getStatus(), logResults);

ChannelResponseCallback2 callback =
            (ChannelResponseCallback2) seq2Callback.get(message.getSeq());

BcosResponseCallback callback = (BcosResponseCallback) seq2Callback.get(message.getSeq());
 --> callback.onResponse(response);
 
ChannelResponseCallback2 callback =
            (ChannelResponseCallback2) seq2Callback.get(message.getSeq());
    if..
        ChannelPush2 push = new ChannelPush2();
        ..
        pushCallback.onPush(push);
    else..
        ..
        callback.onResponse(response);
```

handler：处理连接时、连接后的handler

增加了AMOP的内容
ChannelConnections将加载node.crt node.key的ssl配置SslContext解耦
- initSslContextForConnect()
- initSslContextForListening()

ConnectionCallback，原proxy/Server中的内部类ConnectionCallBack单独拿出来

处理各个DTO类型的Message, ex:

public void onMessage(ChannelHandlerContext ctx, ByteBuf message)
...
if (msg.getType() == ChannelMessageType.BLOCK_NOTIFY.getType()) {
    // new block notify
    ChannelMessage2 channelMessage = new ChannelMessage2(msg);
    channelMessage.readExtra(message);
    channelService.onReceiveBlockNotify(ctx, channelMessage);
    ..
}
// 调用Service中的callback发送通知

其余基本同上...

protocol 规定了channel连接到节点的协议

ChannelProtocol协议内容基类
ChannelMessage(type, error, version, attributeKey等)enum类
- ChannelMessageType
- ChannelMessageError
- EnumChannelProtocolVersion
- EnumSocketChannelAttributeKey
topic相为AMOP
parser
- HeartBeatParser
- BlockNotificationParser

event.filter，新增的模块，事件通知（AMOP？）

EventLogFilter: params, callback, status, channelHandlerContext, filter etc.
- EventLogFilterManager, 与channelService相关，管理Filter状态: start, sendFilter, updateEventLogFilter, addCallback
```
Map<String, EventLogFilter> registerIDToFilter = ConcurrentHashMap
Map<String, Object>  filterIDToCallback = ConcurrentHashMap
```
- EventLogFilterPushStatus push状态 EventLogFilterStatus filter状态
- Response, Exception
- EventLogUserParams
  - validToBlock, validFromToBlock, validAddresses
- EventLogRequestParams 用于request to node
EventLogPushCallBack, EventLogPushWithDecodedCallBack, ServiceEventLogPushCallBack
- onPushEventLog
- transferLogToLogResult
- getDecoder getFilter ...
TopicTools，由string等转为topic中的格式

结合Contract的event触发event log，通过注册EventLogPushCallBack可获得回调

dto

在protocol用到的message, push, callback, request, response等
将EthereumMessage改名为bcosMessage等实体

Proxy和Protocol作用异同

一个是sdk启用一个proxy(Server)，处理消息的上传下发(send/push)

一个是规定消息类型、消息内容等，建立了channel长连接，直接将节点的push在callback中处理

开发中获取callback

sdk只开放了出块blockNotifyCallBack给用户获取，其余的ChannelMessage的callback,或者BcosMessage的callback都拿不到

channel/handler/ConnectionCallBack.java

获取所有的消息通知，

@Override
public void onMessage(ChannelHandlerContext ctx, ByteBuf message) {
    try {
        Message msg = new Message();
        msg.readHeader(message);

        logger.trace(
                "onMessage, seq:{}, type: {}, result: {}",
                msg.getSeq(),
                msg.getType(),
                msg.getResult());

        if (msg.getType() == ChannelMessageType.AMOP_REQUEST.getType()
                || msg.getType() == ChannelMessageType.AMOP_RESPONSE.getType()
                || msg.getType() == ChannelMessageType.AMOP_MULBROADCAST.getType()) {
            ChannelMessage2 channelMessage = new ChannelMessage2(msg);
            channelMessage.readExtra(message);
            channelService.onReceiveChannelMessage2(ctx, channelMessage);
        } else if (msg.getType() == ChannelMessageType.CHANNEL_RPC_REQUEST.getType()) {
            BcosMessage fiscoMessage = new BcosMessage(msg);
            fiscoMessage.readExtra(message);
            channelService.onReceiveEthereumMessage(ctx, fiscoMessage);
        } else if (msg.getType() == ChannelMessageType.CLIENT_HEARTBEAT.getType()) {
            msg.readExtra(message);
            channelService.onReceiveHeartbeat(ctx, msg);
        } else if (msg.getType() == ChannelMessageType.CLIENT_HANDSHAKE.getType()) {
            BcosMessage fiscoMessage = new BcosMessage(msg);
            fiscoMessage.readExtra(message);
            channelService.onReceiveEthereumMessage(ctx, fiscoMessage);
        } else if (msg.getType() == ChannelMessageType.CLIENT_REGISTER_EVENT_LOG.getType()) {
            ChannelMessage2 channelMessage = new ChannelMessage2(msg);
            channelMessage.readExtra(message);
            channelService.onReceiveRegisterEventResponse(ctx, channelMessage);
        } else if (msg.getType() == ChannelMessageType.TRANSACTION_NOTIFY.getType()) {
            BcosMessage fiscoMessage = new BcosMessage(msg);
            fiscoMessage.readExtra(message);
            channelService.onReceiveTransactionMessage(ctx, fiscoMessage);
        } else if (msg.getType() == ChannelMessageType.BLOCK_NOTIFY.getType()) {
            // new block notify
            ChannelMessage2 channelMessage = new ChannelMessage2(msg);
            channelMessage.readExtra(message);
            channelService.onReceiveBlockNotify(ctx, channelMessage);
        } else if (msg.getType() == ChannelMessageType.EVENT_LOG_PUSH.getType()) {
            BcosMessage bcosMessage = new BcosMessage(msg);
            bcosMessage.readExtra(message);
            channelService.onReceiveEventLogPush(ctx, bcosMessage);
        } else if (msg.getType() == ChannelMessageType.REQUEST_TOPICCERT.getType()) {
            logger.info("get generate rand value request data");
            TopicVerifyMessage channelMessage = new TopicVerifyMessage(msg);
            channelMessage.readExtra(message);
            try {
                channelService.checkTopicVerify(ctx, channelMessage);
            } catch (IOException e) {
                logger.error("on receive channel failed");
            }
        } else {
            logger.error("unknown message type:{}", msg.getType());
        }
    } finally {
        message.release();
    }
}