美文网首页
Liquid 使用方法

Liquid 使用方法

作者: 冰冰大象 | 来源:发表于2021-11-23 10:07 被阅读0次

三个节点，测试一下p2p广播功能，点到点消息发送功能和节点发现功能

  1. 场景模拟node1 发送广播，node2和node3 订阅，
  2. node1 点对点 发送数据给node2和node3
  3. node2连接node1 ， node3连接node2，node1发现node3 节点

代码分别如下

node1

/*
Copyright (C) BABEC. All rights reserved.

SPDX-License-Identifier: Apache-2.0
*/

package main

import (
    "chainmaker.org/chainmaker/common/v2/crypto/asym"
    cmTls "chainmaker.org/chainmaker/common/v2/crypto/tls"
    cmx509 "chainmaker.org/chainmaker/common/v2/crypto/x509"
    "chainmaker.org/chainmaker/common/v2/helper"
    "chainmaker.org/chainmaker/net-liquid/core/host"
    "chainmaker.org/chainmaker/net-liquid/core/peer"
    "chainmaker.org/chainmaker/net-liquid/core/protocol"
    "chainmaker.org/chainmaker/net-liquid/core/util"
    "chainmaker.org/chainmaker/net-liquid/discovery/protocoldiscovery"
    nethost "chainmaker.org/chainmaker/net-liquid/host"
    "chainmaker.org/chainmaker/net-liquid/logger"
    "chainmaker.org/chainmaker/net-liquid/pubsub"
    "context"
    "fmt"
    ma "github.com/multiformats/go-multiaddr"
    "strconv"
    "time"
)

const msg = "Hello!My first LIQUID program demo."
const testProtocolID = "/test"
var (
    addrsTcp = []ma.Multiaddr{
        ma.StringCast("/ip4/127.0.0.1/tcp/7081"),
        ma.StringCast("/ip4/127.0.0.1/tcp/7084"),
        ma.StringCast("/ip4/127.0.0.1/tcp/7085"),
    }


    keyPEMs = [][]byte{
        []byte("-----BEGIN EC PRIVATE KEY-----\nMHcCAQEEIDc16StWEUNF0W7iXrT/k0Niy1ZWKbheI2wkjMqJ3WcaoAoGCCqGSM49\nAwEHoUQDQgAEmung19cBCguCaWeyf/nnARHRISTxtaxxf3Zqn+EQxxeYsXqa8oAg\ncalmJaGcmk9kTc0aJwpi7bKvJQdrRg1vFw==\n-----END EC PRIVATE KEY-----\n"),
        []byte("-----BEGIN EC PRIVATE KEY-----\nMHcCAQEEIGLbfX/EvXQtyO+GJvj5rFB0gyTqO3g4gO/gHX2A/BgNoAoGCCqGSM49\nAwEHoUQDQgAE0nVVBXfDQp+EaRbXXfXjn3QG+KYKqApi13+aeNMO7hvS4FlS3B5Z\nMRUCS2oBICwDgLf2q6ef1T5by1u9+IKK7w==\n-----END EC PRIVATE KEY-----\n"),
        []byte("-----BEGIN EC PRIVATE KEY-----\nMHcCAQEEIEUG3DFeXHVr5WjTRfpuSou30zEy2DMmzmyxuSd7F7K9oAoGCCqGSM49\nAwEHoUQDQgAEF+Tvi0fMzJOT9DLS+SFKPIo8kb3ouM54dJ9ibbohpR7FC8Iiid4g\nJ1QNguDjxEmwMV7Q1YiAuighKEmvtb4R+Q==\n-----END EC PRIVATE KEY-----"),
    }

    certPEMs = [][]byte{
        []byte("-----BEGIN CERTIFICATE-----\nMIICcjCCAhegAwIBAgIIFPWUuqixxuAwCgYIKoZIzj0EAwIwZDELMAkGA1UEBhMC\nQ04xEDAOBgNVBAgTB0JlaWppbmcxEDAOBgNVBAcTB0JlaWppbmcxETAPBgNVBAoT\nCGhjLW9yZzU2MQswCQYDVQQLEwJjYTERMA8GA1UEAxMIaGMtb3JnNTYwHhcNMjEx\nMDI4MDgyMzU3WhcNNDExMDIzMDgyMzU3WjBxMQswCQYDVQQGEwJDTjEQMA4GA1UE\nCBMHQmVpamluZzEQMA4GA1UEBxMHQmVpamluZzERMA8GA1UEChMIaGMtb3JnNTYx\nEjAQBgNVBAsTCWNvbnNlbnN1czEXMBUGA1UEAxMOaGMtb3JnNTYtbm9kZTEwWTAT\nBgcqhkjOPQIBBggqhkjOPQMBBwNCAASa6eDX1wEKC4JpZ7J/+ecBEdEhJPG1rHF/\ndmqf4RDHF5ixeprygCBxqWYloZyaT2RNzRonCmLtsq8lB2tGDW8Xo4GlMIGiMA4G\nA1UdDwEB/wQEAwID+DAdBgNVHSUEFjAUBggrBgEFBQcDAgYIKwYBBQUHAwEwKQYD\nVR0OBCIEIOTdkAJ/Fe7PiOnSFe5Fa95mreAwpX2OiK2sB3+Bd86jMCsGA1UdIwQk\nMCKAIKUCNBAgGIS+Af0ucVsMq9ZGS6S2GLlbcDyrtghnqnAnMBkGA1UdEQQSMBCC\nDmhjLW9yZzU2LW5vZGUxMAoGCCqGSM49BAMCA0kAMEYCIQDH+9IoSsBPE33ZRuog\ng9FBCAVj30uuxB08b+eUjbdDGgIhANZ7yABDBk4z9r+gEYtVdFPZdjkEmPa9dYVG\n1u9tUvFy\n-----END CERTIFICATE-----\n"),
        []byte("-----BEGIN CERTIFICATE-----\nMIICcDCCAhegAwIBAgIIJY/rNMZhQFwwCgYIKoZIzj0EAwIwZDELMAkGA1UEBhMC\nQ04xEDAOBgNVBAgTB0JlaWppbmcxEDAOBgNVBAcTB0JlaWppbmcxETAPBgNVBAoT\nCGhjLW9yZzU3MQswCQYDVQQLEwJjYTERMA8GA1UEAxMIaGMtb3JnNTcwHhcNMjEx\nMDI4MDgzMDI5WhcNNDExMDIzMDgzMDI5WjBxMQswCQYDVQQGEwJDTjEQMA4GA1UE\nCBMHQmVpamluZzEQMA4GA1UEBxMHQmVpamluZzERMA8GA1UEChMIaGMtb3JnNTcx\nEjAQBgNVBAsTCWNvbnNlbnN1czEXMBUGA1UEAxMOaGMtb3JnNTctbm9kZTEwWTAT\nBgcqhkjOPQIBBggqhkjOPQMBBwNCAATSdVUFd8NCn4RpFtdd9eOfdAb4pgqoCmLX\nf5p40w7uG9LgWVLcHlkxFQJLagEgLAOAt/arp5/VPlvLW734gorvo4GlMIGiMA4G\nA1UdDwEB/wQEAwID+DAdBgNVHSUEFjAUBggrBgEFBQcDAgYIKwYBBQUHAwEwKQYD\nVR0OBCIEII/jHW2D+EGmFFPJupqBoCubnOwYiEF30VGIZjngZIfwMCsGA1UdIwQk\nMCKAIJ6svKRzGIGG9O0Qh0xEizT54wjh7HivJPiATX5h7ZahMBkGA1UdEQQSMBCC\nDmhjLW9yZzU3LW5vZGUxMAoGCCqGSM49BAMCA0cAMEQCIDES1Cz+VE1uiwSDQYDO\noUD9O51G+a3NMnELPp3GZudUAiBe4jJKk1xPU8jIM6yG5B9q3A7CeASpbbISV0tW\nRDLVYw==\n-----END CERTIFICATE-----\n"),
        []byte("-----BEGIN CERTIFICATE-----\nMIICcjCCAhegAwIBAgIIJFu2YffFkwMwCgYIKoZIzj0EAwIwZDELMAkGA1UEBhMC\nQ04xEDAOBgNVBAgTB0JlaWppbmcxEDAOBgNVBAcTB0JlaWppbmcxETAPBgNVBAoT\nCGhjLW9yZzY1MQswCQYDVQQLEwJjYTERMA8GA1UEAxMIaGMtb3JnNjUwHhcNMjEx\nMTE1MDUzMjQ5WhcNNDExMTEwMDUzMjQ5WjBxMQswCQYDVQQGEwJDTjEQMA4GA1UE\nCBMHQmVpamluZzEQMA4GA1UEBxMHQmVpamluZzERMA8GA1UEChMIaGMtb3JnNjUx\nEjAQBgNVBAsTCWNvbnNlbnN1czEXMBUGA1UEAxMOaGMtb3JnNjUtbm9kZTEwWTAT\nBgcqhkjOPQIBBggqhkjOPQMBBwNCAAQX5O+LR8zMk5P0MtL5IUo8ijyRvei4znh0\nn2JtuiGlHsULwiKJ3iAnVA2C4OPESbAxXtDViIC6KCEoSa+1vhH5o4GlMIGiMA4G\nA1UdDwEB/wQEAwID+DAdBgNVHSUEFjAUBggrBgEFBQcDAgYIKwYBBQUHAwEwKQYD\nVR0OBCIEIGMwbIbdCt+HSL/t2+51nJNAhvH705W8/NnKCj7naO53MCsGA1UdIwQk\nMCKAIIMGhjkBj8VD2+2zAscpecisB5b7dlRdLCJwpce55MSIMBkGA1UdEQQSMBCC\nDmhjLW9yZzY1LW5vZGUxMAoGCCqGSM49BAMCA0kAMEYCIQCSA05P68pGYBCrFoTF\nsPrjx7XN7x1EdFzKb7jnqIc6SQIhAOJYgKkwQQwrXV+yqQJ9XmGRM9I3Ghdvs+Ew\nN4dM580M\n-----END CERTIFICATE-----\n"),
    }

    pidList = []peer.ID{
        "QmVGwL6VXMab7HubwkheZQpJkKKxutqGseJnjg4upWhvDo",
        "Qmav9PUgzrdaLTgJ68XzX9gb4eKLzTonMUpAEsyJRFqV63",
        "QmNQTmmZj8STcDhpKK3izd6q9DT42EuXu7oJyH7Tk1czga",
    }
)


func CreateHostTCP(idx int, seeds map[peer.ID]ma.Multiaddr) (host.Host, error) {
    certPool := cmx509.NewCertPool()
    for i := range certPEMs {
        certPool.AppendCertsFromPEM(certPEMs[i])
    }
    sk, err := asym.PrivateKeyFromPEM(keyPEMs[idx], nil)
    if err != nil {
        return nil, err
    }
    tlsCert, err := cmTls.X509KeyPair(certPEMs[idx], keyPEMs[idx])
    if err != nil {
        return nil, err
    }
    hostCfg := &nethost.HostConfig{
        TlsCfg: &cmTls.Config{
            Certificates:       []cmTls.Certificate{tlsCert},
            InsecureSkipVerify: true,
            ClientAuth:         cmTls.RequireAnyClientCert,
            VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*cmx509.Certificate) error {
                tlsCertBytes := rawCerts[0]
                cert, err := cmx509.ParseCertificate(tlsCertBytes)
                if err != nil {
                    return err
                }
                _, err = cert.Verify(cmx509.VerifyOptions{Roots: certPool})
                if err != nil {
                    return err
                }
                return nil
            },
        },
        LoadPidFunc: func(certificates []*cmx509.Certificate) (peer.ID, error) {
            pid, err := helper.GetLibp2pPeerIdFromCertDer(certificates[0].Raw)
            if err != nil {
                return "", err
            }
            return peer.ID(pid), err
        },
        SendStreamPoolInitSize:    10,
        SendStreamPoolCap:         50,
        PeerReceiveStreamMaxCount: 100,
        ListenAddresses:           []ma.Multiaddr{addrsTcp[idx]},
        DirectPeers:               seeds,
        MsgCompress:               false,
        Insecurity:                false,
        PrivateKey:                sk,
    }

    return hostCfg.NewHost(nethost.TcpNetwork, context.Background(), logger.NewLogPrinter("HOST"+strconv.Itoa(idx)))
}

func TestHostTCP() {
    // create host1 pidList[1]: ma.Join(addr2TargetTcp, ma.StringCast("/p2p/"+pidList[1].ToString()))
    host1, err := CreateHostTCP(0, map[peer.ID]ma.Multiaddr{})
    // register notifee
    notifyBundle := &host.NotifieeBundle{
        PeerConnectedFunc: func(id peer.ID) {
            fmt.Printf("节点已连接，节点ID：%s\n", id)
        },
        PeerDisconnectedFunc: func(id peer.ID) {
            fmt.Printf("节点已断开，节点ID：%s\n", id)
        },
        PeerProtocolSupportedFunc: func(protocolID protocol.ID, pid peer.ID) {
            fmt.Printf("节点%s新支持协议%s\n", pid, protocolID)
        },
        PeerProtocolUnsupportedFunc: func(protocolID protocol.ID, pid peer.ID) {
            fmt.Printf("节点%s取消支持协议%s\n", pid, protocolID)
        },
    }
    host1.Notify(notifyBundle)
    ps := pubsub.NewChainPubSub("c1635410037173", logger.NewLogPrinter("PubSub"))
    err = ps.AttachHost(host1)
    if err != nil {
        // do something
        panic(err)
    }
    // start hosts
    err = host1.Start()

    err = host1.RegisterMsgPayloadHandler(testProtocolID, func(senderPID peer.ID, msgPayload []byte) {
        fmt.Println(string(msgPayload))
    })


    testTopic := "topic"
    go func() {
        i:=0
        for  {
            i++
            // host1 send msg to host2
            time.Sleep(time.Second)
            err = host1.SendMsg(testProtocolID, pidList[1], []byte(msg))
            err = host1.SendMsg(testProtocolID, pidList[2], []byte(msg))
            ps.Publish(testTopic,[]byte("广播的消息"+strconv.Itoa(i)))
        }
    }()
    // 节点发现服务
    // 10. 开启节点发现服务
    log := logger.NewLogPrinter("TEST")
    discoveryService, err := protocoldiscovery.NewProtocolBasedDiscovery(host1, protocoldiscovery.WithLogger(log))
    if err != nil {
        // do something
        panic(err)
    }
    ctx := context.Background()
    // 11. 宣布自己支持的服务
    err = discoveryService.Announce(ctx, testProtocolID)
    if err != nil {
        // do something
        panic(err)
    }

    // 12. 搜寻其他节点
    findC, err := discoveryService.FindPeers(ctx, testProtocolID)
    if err != nil {
        // do something
        panic(err)
    }
    go listenFindingC(ctx,0,host1,findC)
    select {

    }

}

func listenFindingC(ctx context.Context, idx int, h host.Host, c <-chan ma.Multiaddr) {
    for {
        select {
        case <-ctx.Done():
            return
        case ai := <-c:
            fmt.Printf("发现新节点，地址：%s\n", ai.String())
            addr, pid := util.GetNetAddrAndPidFromNormalMultiAddr(ai)
            if pid == "" {
                fmt.Errorf("[Discovery%d] peer id not contains in addr", idx)
                continue
            }
            if h.ConnMgr().PeerCount() >= h.ConnMgr().MaxPeerCountAllowed() || h.ID() == pid || h.ConnMgr().IsConnected(pid) {
                continue
            }
            fmt.Printf("[Discovery%d] find new peer.(pid: %s, addr: %s)", idx, pid, addr.String())
            _, err := h.Dial(ai)
            if err!=nil{
                fmt.Errorf("连接出现错误")
                continue
            }

        }
    }
}

node 2

/*
Copyright (C) BABEC. All rights reserved.

SPDX-License-Identifier: Apache-2.0
*/

package main

import (
    "chainmaker.org/chainmaker/common/v2/crypto/asym"
    cmTls "chainmaker.org/chainmaker/common/v2/crypto/tls"
    cmx509 "chainmaker.org/chainmaker/common/v2/crypto/x509"
    "chainmaker.org/chainmaker/common/v2/helper"
    "chainmaker.org/chainmaker/net-liquid/core/handler"
    "chainmaker.org/chainmaker/net-liquid/core/host"
    "chainmaker.org/chainmaker/net-liquid/core/peer"
    "chainmaker.org/chainmaker/net-liquid/core/protocol"
    "chainmaker.org/chainmaker/net-liquid/core/util"
    "chainmaker.org/chainmaker/net-liquid/discovery/protocoldiscovery"
    nethost "chainmaker.org/chainmaker/net-liquid/host"
    "chainmaker.org/chainmaker/net-liquid/logger"
    "chainmaker.org/chainmaker/net-liquid/pubsub"
    "context"
    "fmt"
    ma "github.com/multiformats/go-multiaddr"
    "strconv"
)

const msg = "Hello!My first LIQUID program demo."
const testProtocolID = "/test"
var (
    addrsTcp = []ma.Multiaddr{
        ma.StringCast("/ip4/127.0.0.1/tcp/7081"),
        ma.StringCast("/ip4/127.0.0.1/tcp/7084"),
        ma.StringCast("/ip4/127.0.0.1/tcp/7085"),
    }


    keyPEMs = [][]byte{
        []byte("-----BEGIN EC PRIVATE KEY-----\nMHcCAQEEIDc16StWEUNF0W7iXrT/k0Niy1ZWKbheI2wkjMqJ3WcaoAoGCCqGSM49\nAwEHoUQDQgAEmung19cBCguCaWeyf/nnARHRISTxtaxxf3Zqn+EQxxeYsXqa8oAg\ncalmJaGcmk9kTc0aJwpi7bKvJQdrRg1vFw==\n-----END EC PRIVATE KEY-----\n"),
        []byte("-----BEGIN EC PRIVATE KEY-----\nMHcCAQEEIGLbfX/EvXQtyO+GJvj5rFB0gyTqO3g4gO/gHX2A/BgNoAoGCCqGSM49\nAwEHoUQDQgAE0nVVBXfDQp+EaRbXXfXjn3QG+KYKqApi13+aeNMO7hvS4FlS3B5Z\nMRUCS2oBICwDgLf2q6ef1T5by1u9+IKK7w==\n-----END EC PRIVATE KEY-----\n"),
        []byte("-----BEGIN EC PRIVATE KEY-----\nMHcCAQEEIEUG3DFeXHVr5WjTRfpuSou30zEy2DMmzmyxuSd7F7K9oAoGCCqGSM49\nAwEHoUQDQgAEF+Tvi0fMzJOT9DLS+SFKPIo8kb3ouM54dJ9ibbohpR7FC8Iiid4g\nJ1QNguDjxEmwMV7Q1YiAuighKEmvtb4R+Q==\n-----END EC PRIVATE KEY-----"),
    }

    certPEMs = [][]byte{
        []byte("-----BEGIN CERTIFICATE-----\nMIICcjCCAhegAwIBAgIIFPWUuqixxuAwCgYIKoZIzj0EAwIwZDELMAkGA1UEBhMC\nQ04xEDAOBgNVBAgTB0JlaWppbmcxEDAOBgNVBAcTB0JlaWppbmcxETAPBgNVBAoT\nCGhjLW9yZzU2MQswCQYDVQQLEwJjYTERMA8GA1UEAxMIaGMtb3JnNTYwHhcNMjEx\nMDI4MDgyMzU3WhcNNDExMDIzMDgyMzU3WjBxMQswCQYDVQQGEwJDTjEQMA4GA1UE\nCBMHQmVpamluZzEQMA4GA1UEBxMHQmVpamluZzERMA8GA1UEChMIaGMtb3JnNTYx\nEjAQBgNVBAsTCWNvbnNlbnN1czEXMBUGA1UEAxMOaGMtb3JnNTYtbm9kZTEwWTAT\nBgcqhkjOPQIBBggqhkjOPQMBBwNCAASa6eDX1wEKC4JpZ7J/+ecBEdEhJPG1rHF/\ndmqf4RDHF5ixeprygCBxqWYloZyaT2RNzRonCmLtsq8lB2tGDW8Xo4GlMIGiMA4G\nA1UdDwEB/wQEAwID+DAdBgNVHSUEFjAUBggrBgEFBQcDAgYIKwYBBQUHAwEwKQYD\nVR0OBCIEIOTdkAJ/Fe7PiOnSFe5Fa95mreAwpX2OiK2sB3+Bd86jMCsGA1UdIwQk\nMCKAIKUCNBAgGIS+Af0ucVsMq9ZGS6S2GLlbcDyrtghnqnAnMBkGA1UdEQQSMBCC\nDmhjLW9yZzU2LW5vZGUxMAoGCCqGSM49BAMCA0kAMEYCIQDH+9IoSsBPE33ZRuog\ng9FBCAVj30uuxB08b+eUjbdDGgIhANZ7yABDBk4z9r+gEYtVdFPZdjkEmPa9dYVG\n1u9tUvFy\n-----END CERTIFICATE-----\n"),
        []byte("-----BEGIN CERTIFICATE-----\nMIICcDCCAhegAwIBAgIIJY/rNMZhQFwwCgYIKoZIzj0EAwIwZDELMAkGA1UEBhMC\nQ04xEDAOBgNVBAgTB0JlaWppbmcxEDAOBgNVBAcTB0JlaWppbmcxETAPBgNVBAoT\nCGhjLW9yZzU3MQswCQYDVQQLEwJjYTERMA8GA1UEAxMIaGMtb3JnNTcwHhcNMjEx\nMDI4MDgzMDI5WhcNNDExMDIzMDgzMDI5WjBxMQswCQYDVQQGEwJDTjEQMA4GA1UE\nCBMHQmVpamluZzEQMA4GA1UEBxMHQmVpamluZzERMA8GA1UEChMIaGMtb3JnNTcx\nEjAQBgNVBAsTCWNvbnNlbnN1czEXMBUGA1UEAxMOaGMtb3JnNTctbm9kZTEwWTAT\nBgcqhkjOPQIBBggqhkjOPQMBBwNCAATSdVUFd8NCn4RpFtdd9eOfdAb4pgqoCmLX\nf5p40w7uG9LgWVLcHlkxFQJLagEgLAOAt/arp5/VPlvLW734gorvo4GlMIGiMA4G\nA1UdDwEB/wQEAwID+DAdBgNVHSUEFjAUBggrBgEFBQcDAgYIKwYBBQUHAwEwKQYD\nVR0OBCIEII/jHW2D+EGmFFPJupqBoCubnOwYiEF30VGIZjngZIfwMCsGA1UdIwQk\nMCKAIJ6svKRzGIGG9O0Qh0xEizT54wjh7HivJPiATX5h7ZahMBkGA1UdEQQSMBCC\nDmhjLW9yZzU3LW5vZGUxMAoGCCqGSM49BAMCA0cAMEQCIDES1Cz+VE1uiwSDQYDO\noUD9O51G+a3NMnELPp3GZudUAiBe4jJKk1xPU8jIM6yG5B9q3A7CeASpbbISV0tW\nRDLVYw==\n-----END CERTIFICATE-----\n"),
        []byte("-----BEGIN CERTIFICATE-----\nMIICcjCCAhegAwIBAgIIJFu2YffFkwMwCgYIKoZIzj0EAwIwZDELMAkGA1UEBhMC\nQ04xEDAOBgNVBAgTB0JlaWppbmcxEDAOBgNVBAcTB0JlaWppbmcxETAPBgNVBAoT\nCGhjLW9yZzY1MQswCQYDVQQLEwJjYTERMA8GA1UEAxMIaGMtb3JnNjUwHhcNMjEx\nMTE1MDUzMjQ5WhcNNDExMTEwMDUzMjQ5WjBxMQswCQYDVQQGEwJDTjEQMA4GA1UE\nCBMHQmVpamluZzEQMA4GA1UEBxMHQmVpamluZzERMA8GA1UEChMIaGMtb3JnNjUx\nEjAQBgNVBAsTCWNvbnNlbnN1czEXMBUGA1UEAxMOaGMtb3JnNjUtbm9kZTEwWTAT\nBgcqhkjOPQIBBggqhkjOPQMBBwNCAAQX5O+LR8zMk5P0MtL5IUo8ijyRvei4znh0\nn2JtuiGlHsULwiKJ3iAnVA2C4OPESbAxXtDViIC6KCEoSa+1vhH5o4GlMIGiMA4G\nA1UdDwEB/wQEAwID+DAdBgNVHSUEFjAUBggrBgEFBQcDAgYIKwYBBQUHAwEwKQYD\nVR0OBCIEIGMwbIbdCt+HSL/t2+51nJNAhvH705W8/NnKCj7naO53MCsGA1UdIwQk\nMCKAIIMGhjkBj8VD2+2zAscpecisB5b7dlRdLCJwpce55MSIMBkGA1UdEQQSMBCC\nDmhjLW9yZzY1LW5vZGUxMAoGCCqGSM49BAMCA0kAMEYCIQCSA05P68pGYBCrFoTF\nsPrjx7XN7x1EdFzKb7jnqIc6SQIhAOJYgKkwQQwrXV+yqQJ9XmGRM9I3Ghdvs+Ew\nN4dM580M\n-----END CERTIFICATE-----\n"),
    }

    pidList = []peer.ID{
        "QmVGwL6VXMab7HubwkheZQpJkKKxutqGseJnjg4upWhvDo",
        "Qmav9PUgzrdaLTgJ68XzX9gb4eKLzTonMUpAEsyJRFqV63",
        "QmNQTmmZj8STcDhpKK3izd6q9DT42EuXu7oJyH7Tk1czga",
    }
)

func CreateHostTCP(idx int, seeds map[peer.ID]ma.Multiaddr) (host.Host, error) {
    certPool := cmx509.NewCertPool()
    for i := range certPEMs {
        certPool.AppendCertsFromPEM(certPEMs[i])
    }
    sk, err := asym.PrivateKeyFromPEM(keyPEMs[idx], nil)
    if err != nil {
        return nil, err
    }
    tlsCert, err := cmTls.X509KeyPair(certPEMs[idx], keyPEMs[idx])
    if err != nil {
        return nil, err
    }
    hostCfg := &nethost.HostConfig{
        TlsCfg: &cmTls.Config{
            Certificates:       []cmTls.Certificate{tlsCert},
            InsecureSkipVerify: true,
            ClientAuth:         cmTls.RequireAnyClientCert,
            VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*cmx509.Certificate) error {
                tlsCertBytes := rawCerts[0]
                cert, err := cmx509.ParseCertificate(tlsCertBytes)
                if err != nil {
                    return err
                }
                _, err = cert.Verify(cmx509.VerifyOptions{Roots: certPool})
                if err != nil {
                    return err
                }
                return nil
            },
        },
        LoadPidFunc: func(certificates []*cmx509.Certificate) (peer.ID, error) {
            pid, err := helper.GetLibp2pPeerIdFromCertDer(certificates[0].Raw)
            if err != nil {
                return "", err
            }
            return peer.ID(pid), err
        },
        SendStreamPoolInitSize:    10,
        SendStreamPoolCap:         50,
        PeerReceiveStreamMaxCount: 100,
        ListenAddresses:           []ma.Multiaddr{addrsTcp[idx]},
        DirectPeers:               seeds,
        MsgCompress:               false,
        Insecurity:                false,
        PrivateKey:                sk,
    }
    return hostCfg.NewHost(nethost.TcpNetwork, context.Background(), logger.NewLogPrinter("HOST"+strconv.Itoa(idx)))
}
func TestHostTCP() {
    // create host2 pidList[0]: ma.Join(addrsTcp[0], ma.StringCast("/p2p/"+pidList[0].ToString()))
    host1, _ := CreateHostTCP(1, map[peer.ID]ma.Multiaddr{pidList[0]: ma.Join(addrsTcp[0], ma.StringCast("/p2p/"+pidList[0].ToString()))})

    notifyBundle := &host.NotifieeBundle{
        PeerConnectedFunc: func(id peer.ID) {
            fmt.Printf("节点已连接，节点ID：%s\n", id)
        },
        PeerDisconnectedFunc: func(id peer.ID) {
            fmt.Printf("节点已断开，节点ID：%s\n", id)
        },
        PeerProtocolSupportedFunc: func(protocolID protocol.ID, pid peer.ID) {
            fmt.Printf("节点%s新支持协议%s\n", pid, protocolID)
        },
        PeerProtocolUnsupportedFunc: func(protocolID protocol.ID, pid peer.ID) {
            fmt.Printf("节点%s取消支持协议%s\n", pid, protocolID)
        },
    }
    host1.Notify(notifyBundle)

    ps := pubsub.NewChainPubSub("c1635410037173", logger.NewLogPrinter("PubSub"))
    err := ps.AttachHost(host1)
    if err != nil {
        // do something
        panic(err)
    }
    testTopic := "topic"
    var subHandler handler.SubMsgHandler = func(publisher peer.ID, topic string, msg []byte) {
        fmt.Printf("收到节点%s向频道%s发布的消息：%s\n", publisher, topic, string(msg))
    }
    ps.Subscribe(testTopic, subHandler)
    // start hosts
    _ = host1.Start()
    // wait for connection established between host1 and host2

    // register msg payload handler
    _ = host1.RegisterMsgPayloadHandler(testProtocolID, func(senderPID peer.ID, msgPayload []byte) {
        fmt.Println("收到Message"+string(msgPayload))
    })

    // 节点发现服务
    // 10. 开启节点发现服务
    log := logger.NewLogPrinter("TEST")
    discoveryService, err := protocoldiscovery.NewProtocolBasedDiscovery(host1, protocoldiscovery.WithLogger(log))
    if err != nil {
        // do something
        panic(err)
    }
    ctx := context.Background()
    // 11. 宣布自己支持的服务
    err = discoveryService.Announce(ctx, testProtocolID)
    if err != nil {
        // do something
        panic(err)
    }

    // 12. 搜寻其他节点
    findC, err := discoveryService.FindPeers(ctx, testProtocolID)
    if err != nil {
        // do something
        panic(err)
    }
    go listenFindingC(ctx,1,host1,findC)
    select {

    }

}

func listenFindingC(ctx context.Context, idx int, h host.Host, c <-chan ma.Multiaddr) {
    for {
        select {
        case <-ctx.Done():
            return
        case ai := <-c:
            fmt.Printf("发现新节点，地址：%s\n", ai.String())
            addr, pid := util.GetNetAddrAndPidFromNormalMultiAddr(ai)
            if pid == "" {
                fmt.Errorf("[Discovery%d] peer id not contains in addr", idx)
                continue
            }
            if h.ConnMgr().PeerCount() >= h.ConnMgr().MaxPeerCountAllowed() || h.ID() == pid || h.ConnMgr().IsConnected(pid) {
                continue
            }
            fmt.Printf("[Discovery%d] find new peer.(pid: %s, addr: %s)", idx, pid, addr.String())
            _, _ = h.Dial(ai)
        }
    }
}

node3

/*
Copyright (C) BABEC. All rights reserved.

SPDX-License-Identifier: Apache-2.0
*/

package main

import (
    "chainmaker.org/chainmaker/common/v2/crypto/asym"
    cmTls "chainmaker.org/chainmaker/common/v2/crypto/tls"
    cmx509 "chainmaker.org/chainmaker/common/v2/crypto/x509"
    "chainmaker.org/chainmaker/common/v2/helper"
    "chainmaker.org/chainmaker/net-liquid/core/handler"
    "chainmaker.org/chainmaker/net-liquid/core/host"
    "chainmaker.org/chainmaker/net-liquid/core/peer"
    "chainmaker.org/chainmaker/net-liquid/core/protocol"
    "chainmaker.org/chainmaker/net-liquid/core/util"
    "chainmaker.org/chainmaker/net-liquid/discovery/protocoldiscovery"
    nethost "chainmaker.org/chainmaker/net-liquid/host"
    "chainmaker.org/chainmaker/net-liquid/logger"
    "chainmaker.org/chainmaker/net-liquid/pubsub"
    "context"
    "fmt"
    ma "github.com/multiformats/go-multiaddr"
    "strconv"
)

const msg = "Hello!My first LIQUID program demo."
const testProtocolID = "/test"
var (
    addrsTcp = []ma.Multiaddr{
        ma.StringCast("/ip4/127.0.0.1/tcp/7081"),
        ma.StringCast("/ip4/127.0.0.1/tcp/7084"),
        ma.StringCast("/ip4/127.0.0.1/tcp/7085"),
    }


    keyPEMs = [][]byte{
        []byte("-----BEGIN EC PRIVATE KEY-----\nMHcCAQEEIDc16StWEUNF0W7iXrT/k0Niy1ZWKbheI2wkjMqJ3WcaoAoGCCqGSM49\nAwEHoUQDQgAEmung19cBCguCaWeyf/nnARHRISTxtaxxf3Zqn+EQxxeYsXqa8oAg\ncalmJaGcmk9kTc0aJwpi7bKvJQdrRg1vFw==\n-----END EC PRIVATE KEY-----\n"),
        []byte("-----BEGIN EC PRIVATE KEY-----\nMHcCAQEEIGLbfX/EvXQtyO+GJvj5rFB0gyTqO3g4gO/gHX2A/BgNoAoGCCqGSM49\nAwEHoUQDQgAE0nVVBXfDQp+EaRbXXfXjn3QG+KYKqApi13+aeNMO7hvS4FlS3B5Z\nMRUCS2oBICwDgLf2q6ef1T5by1u9+IKK7w==\n-----END EC PRIVATE KEY-----\n"),
        []byte("-----BEGIN EC PRIVATE KEY-----\nMHcCAQEEIEUG3DFeXHVr5WjTRfpuSou30zEy2DMmzmyxuSd7F7K9oAoGCCqGSM49\nAwEHoUQDQgAEF+Tvi0fMzJOT9DLS+SFKPIo8kb3ouM54dJ9ibbohpR7FC8Iiid4g\nJ1QNguDjxEmwMV7Q1YiAuighKEmvtb4R+Q==\n-----END EC PRIVATE KEY-----"),
    }

    certPEMs = [][]byte{
        []byte("-----BEGIN CERTIFICATE-----\nMIICcjCCAhegAwIBAgIIFPWUuqixxuAwCgYIKoZIzj0EAwIwZDELMAkGA1UEBhMC\nQ04xEDAOBgNVBAgTB0JlaWppbmcxEDAOBgNVBAcTB0JlaWppbmcxETAPBgNVBAoT\nCGhjLW9yZzU2MQswCQYDVQQLEwJjYTERMA8GA1UEAxMIaGMtb3JnNTYwHhcNMjEx\nMDI4MDgyMzU3WhcNNDExMDIzMDgyMzU3WjBxMQswCQYDVQQGEwJDTjEQMA4GA1UE\nCBMHQmVpamluZzEQMA4GA1UEBxMHQmVpamluZzERMA8GA1UEChMIaGMtb3JnNTYx\nEjAQBgNVBAsTCWNvbnNlbnN1czEXMBUGA1UEAxMOaGMtb3JnNTYtbm9kZTEwWTAT\nBgcqhkjOPQIBBggqhkjOPQMBBwNCAASa6eDX1wEKC4JpZ7J/+ecBEdEhJPG1rHF/\ndmqf4RDHF5ixeprygCBxqWYloZyaT2RNzRonCmLtsq8lB2tGDW8Xo4GlMIGiMA4G\nA1UdDwEB/wQEAwID+DAdBgNVHSUEFjAUBggrBgEFBQcDAgYIKwYBBQUHAwEwKQYD\nVR0OBCIEIOTdkAJ/Fe7PiOnSFe5Fa95mreAwpX2OiK2sB3+Bd86jMCsGA1UdIwQk\nMCKAIKUCNBAgGIS+Af0ucVsMq9ZGS6S2GLlbcDyrtghnqnAnMBkGA1UdEQQSMBCC\nDmhjLW9yZzU2LW5vZGUxMAoGCCqGSM49BAMCA0kAMEYCIQDH+9IoSsBPE33ZRuog\ng9FBCAVj30uuxB08b+eUjbdDGgIhANZ7yABDBk4z9r+gEYtVdFPZdjkEmPa9dYVG\n1u9tUvFy\n-----END CERTIFICATE-----\n"),
        []byte("-----BEGIN CERTIFICATE-----\nMIICcDCCAhegAwIBAgIIJY/rNMZhQFwwCgYIKoZIzj0EAwIwZDELMAkGA1UEBhMC\nQ04xEDAOBgNVBAgTB0JlaWppbmcxEDAOBgNVBAcTB0JlaWppbmcxETAPBgNVBAoT\nCGhjLW9yZzU3MQswCQYDVQQLEwJjYTERMA8GA1UEAxMIaGMtb3JnNTcwHhcNMjEx\nMDI4MDgzMDI5WhcNNDExMDIzMDgzMDI5WjBxMQswCQYDVQQGEwJDTjEQMA4GA1UE\nCBMHQmVpamluZzEQMA4GA1UEBxMHQmVpamluZzERMA8GA1UEChMIaGMtb3JnNTcx\nEjAQBgNVBAsTCWNvbnNlbnN1czEXMBUGA1UEAxMOaGMtb3JnNTctbm9kZTEwWTAT\nBgcqhkjOPQIBBggqhkjOPQMBBwNCAATSdVUFd8NCn4RpFtdd9eOfdAb4pgqoCmLX\nf5p40w7uG9LgWVLcHlkxFQJLagEgLAOAt/arp5/VPlvLW734gorvo4GlMIGiMA4G\nA1UdDwEB/wQEAwID+DAdBgNVHSUEFjAUBggrBgEFBQcDAgYIKwYBBQUHAwEwKQYD\nVR0OBCIEII/jHW2D+EGmFFPJupqBoCubnOwYiEF30VGIZjngZIfwMCsGA1UdIwQk\nMCKAIJ6svKRzGIGG9O0Qh0xEizT54wjh7HivJPiATX5h7ZahMBkGA1UdEQQSMBCC\nDmhjLW9yZzU3LW5vZGUxMAoGCCqGSM49BAMCA0cAMEQCIDES1Cz+VE1uiwSDQYDO\noUD9O51G+a3NMnELPp3GZudUAiBe4jJKk1xPU8jIM6yG5B9q3A7CeASpbbISV0tW\nRDLVYw==\n-----END CERTIFICATE-----\n"),
        []byte("-----BEGIN CERTIFICATE-----\nMIICcjCCAhegAwIBAgIIJFu2YffFkwMwCgYIKoZIzj0EAwIwZDELMAkGA1UEBhMC\nQ04xEDAOBgNVBAgTB0JlaWppbmcxEDAOBgNVBAcTB0JlaWppbmcxETAPBgNVBAoT\nCGhjLW9yZzY1MQswCQYDVQQLEwJjYTERMA8GA1UEAxMIaGMtb3JnNjUwHhcNMjEx\nMTE1MDUzMjQ5WhcNNDExMTEwMDUzMjQ5WjBxMQswCQYDVQQGEwJDTjEQMA4GA1UE\nCBMHQmVpamluZzEQMA4GA1UEBxMHQmVpamluZzERMA8GA1UEChMIaGMtb3JnNjUx\nEjAQBgNVBAsTCWNvbnNlbnN1czEXMBUGA1UEAxMOaGMtb3JnNjUtbm9kZTEwWTAT\nBgcqhkjOPQIBBggqhkjOPQMBBwNCAAQX5O+LR8zMk5P0MtL5IUo8ijyRvei4znh0\nn2JtuiGlHsULwiKJ3iAnVA2C4OPESbAxXtDViIC6KCEoSa+1vhH5o4GlMIGiMA4G\nA1UdDwEB/wQEAwID+DAdBgNVHSUEFjAUBggrBgEFBQcDAgYIKwYBBQUHAwEwKQYD\nVR0OBCIEIGMwbIbdCt+HSL/t2+51nJNAhvH705W8/NnKCj7naO53MCsGA1UdIwQk\nMCKAIIMGhjkBj8VD2+2zAscpecisB5b7dlRdLCJwpce55MSIMBkGA1UdEQQSMBCC\nDmhjLW9yZzY1LW5vZGUxMAoGCCqGSM49BAMCA0kAMEYCIQCSA05P68pGYBCrFoTF\nsPrjx7XN7x1EdFzKb7jnqIc6SQIhAOJYgKkwQQwrXV+yqQJ9XmGRM9I3Ghdvs+Ew\nN4dM580M\n-----END CERTIFICATE-----\n"),
    }

    pidList = []peer.ID{
        "QmVGwL6VXMab7HubwkheZQpJkKKxutqGseJnjg4upWhvDo",
        "Qmav9PUgzrdaLTgJ68XzX9gb4eKLzTonMUpAEsyJRFqV63",
        "QmNQTmmZj8STcDhpKK3izd6q9DT42EuXu7oJyH7Tk1czga",
    }
)

func CreateHostTCP(idx int, seeds map[peer.ID]ma.Multiaddr) (host.Host, error) {
    certPool := cmx509.NewCertPool()
    for i := range certPEMs {
        certPool.AppendCertsFromPEM(certPEMs[i])
    }
    sk, err := asym.PrivateKeyFromPEM(keyPEMs[idx], nil)
    if err != nil {
        return nil, err
    }
    tlsCert, err := cmTls.X509KeyPair(certPEMs[idx], keyPEMs[idx])
    if err != nil {
        return nil, err
    }
    hostCfg := &nethost.HostConfig{
        TlsCfg: &cmTls.Config{
            Certificates:       []cmTls.Certificate{tlsCert},
            InsecureSkipVerify: true,
            ClientAuth:         cmTls.RequireAnyClientCert,
            VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*cmx509.Certificate) error {
                tlsCertBytes := rawCerts[0]
                cert, err := cmx509.ParseCertificate(tlsCertBytes)
                if err != nil {
                    return err
                }
                _, err = cert.Verify(cmx509.VerifyOptions{Roots: certPool})
                if err != nil {
                    return err
                }
                return nil
            },
        },
        LoadPidFunc: func(certificates []*cmx509.Certificate) (peer.ID, error) {
            pid, err := helper.GetLibp2pPeerIdFromCertDer(certificates[0].Raw)
            if err != nil {
                return "", err
            }
            return peer.ID(pid), err
        },
        SendStreamPoolInitSize:    10,
        SendStreamPoolCap:         50,
        PeerReceiveStreamMaxCount: 100,
        ListenAddresses:           []ma.Multiaddr{addrsTcp[idx]},
        DirectPeers:               seeds,
        MsgCompress:               false,
        Insecurity:                false,
        PrivateKey:                sk,
    }
    return hostCfg.NewHost(nethost.TcpNetwork, context.Background(), logger.NewLogPrinter("HOST"+strconv.Itoa(idx)))
}
func TestHostTCP() {
    // create host2 pidList[0]: ma.Join(addrsTcp[0], ma.StringCast("/p2p/"+pidList[0].ToString()))
    host1, _ := CreateHostTCP(2, map[peer.ID]ma.Multiaddr{pidList[1]: ma.Join(addrsTcp[1], ma.StringCast("/p2p/"+pidList[1].ToString()))})

    notifyBundle := &host.NotifieeBundle{
        PeerConnectedFunc: func(id peer.ID) {
            fmt.Printf("节点已连接，节点ID：%s\n", id)
        },
        PeerDisconnectedFunc: func(id peer.ID) {
            fmt.Printf("节点已断开，节点ID：%s\n", id)
        },
        PeerProtocolSupportedFunc: func(protocolID protocol.ID, pid peer.ID) {
            fmt.Printf("节点%s新支持协议%s\n", pid, protocolID)
        },
        PeerProtocolUnsupportedFunc: func(protocolID protocol.ID, pid peer.ID) {
            fmt.Printf("节点%s取消支持协议%s\n", pid, protocolID)
        },
    }
    host1.Notify(notifyBundle)

    ps := pubsub.NewChainPubSub("c1635410037173", logger.NewLogPrinter("PubSub"))
    err := ps.AttachHost(host1)
    if err != nil {
        // do something
        panic(err)
    }
    testTopic := "topic"
    var subHandler handler.SubMsgHandler = func(publisher peer.ID, topic string, msg []byte) {
        fmt.Printf("收到节点%s向频道%s发布的消息：%s\n", publisher, topic, string(msg))
    }
    ps.Subscribe(testTopic, subHandler)
    // start hosts
    _ = host1.Start()
    // wait for connection established between host1 and host2

    // register msg payload handler
    _ = host1.RegisterMsgPayloadHandler(testProtocolID, func(senderPID peer.ID, msgPayload []byte) {
        fmt.Println("收到Message"+string(msgPayload))
    })

    // 节点发现服务
    // 10. 开启节点发现服务
    log := logger.NewLogPrinter("TEST")
    discoveryService, err := protocoldiscovery.NewProtocolBasedDiscovery(host1, protocoldiscovery.WithLogger(log))
    if err != nil {
        // do something
        panic(err)
    }
    ctx := context.Background()
    // 11. 宣布自己支持的服务
    err = discoveryService.Announce(ctx, testProtocolID)
    if err != nil {
        // do something
        panic(err)
    }

    // 12. 搜寻其他节点
    findC, err := discoveryService.FindPeers(ctx, testProtocolID)
    if err != nil {
        // do something
        panic(err)
    }
    go listenFindingC(ctx,2,host1,findC)
    select {

    }

}

func listenFindingC(ctx context.Context, idx int, h host.Host, c <-chan ma.Multiaddr) {
    for {
        select {
        case <-ctx.Done():
            return
        case ai := <-c:
            fmt.Printf("发现新节点，地址：%s\n", ai.String())
            addr, pid := util.GetNetAddrAndPidFromNormalMultiAddr(ai)
            if pid == "" {
                fmt.Errorf("[Discovery%d] peer id not contains in addr", idx)
                continue
            }
            if h.ConnMgr().PeerCount() >= h.ConnMgr().MaxPeerCountAllowed() || h.ID() == pid || h.ConnMgr().IsConnected(pid) {
                continue
            }
            fmt.Printf("[Discovery%d] find new peer.(pid: %s, addr: %s)", idx, pid, addr.String())
            _, _ = h.Dial(ai)
        }
    }
}

相关文章

  • Liquid 使用方法

    三个节点，测试一下p2p广播功能，点到点消息发送功能和节点发现功能 场景模拟node1 发送广播，node2和no...

  • 区块链周刊(2018-10-21)

    新闻 Liquid Network(比特币侧链)上线 我们很高兴地宣布，Liquid Network已经上线。作为...

  • Jekyll Liquid使用小问题: {% raw %} 引发

    今天写了篇Liquid的笔记. 由于用到大量Liquid语法, 因此markdown文件内就使用了{% raw %...

  • Liquid Modernity

    我很久没有写东西了，上学那会叫写作文，大学毕业后就几乎没有糟蹋过一张纸！从13年毕业到某知名开发商做房地产销售，使...

  • Liquid语法

    引用官方文档 一、基本 １、介绍 liquid 代码可以被分类为对象，标签和过滤代码 对象对象表示页面中将要显示的...

  • liquid love

    渴望与人相系，却更恐惧被就此绑死。害怕孤单，所以找人同行，但新的不安却从别处涌现。在液态的爱里，我们一次一次地聚合...

  • liquid crystal

    life is brief. liquid 液体 water . oil and milk are liquids...

  • m端添加搜索图标

    1.在theme.scss.liquid末尾空白处加上 2.在header.liquid页面中 schema前面...

  • prestige主题移动端头部添加搜索框

    1。在header.liquid搜索/cart然后添加下面的代码 同样在header.liquid搜索 /sear...

  • Liquid 语言基础

    Liquid一门为灵活的 web 应用而设计的面向客户的模板语言。 原文链接 简介 什么是 Liquid ？ Li...

网友评论

      本文标题：Liquid 使用方法

      本文链接：https://www.haomeiwen.com/subject/oxfwtrtx.html

      延伸阅读

      深度阅读

      • 您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！

      栏目导航

      热点阅读

      关于我们|服务条款|联系我们|Liquid 使用方法|投稿指南|网站地图|RSS订阅|排版工具|手机版

      提供经典美文摘抄,优美散文欣赏,现代诗歌精选,短篇小说,心情随笔,表白情书范文,故事会在线阅读欣赏

      Copyright © 2014-2023 Haomeiwen.com All Rights Reserved. 好美文阅读网 版权所有

      备案信息：桂公网安备 45052102000051号 · 桂ICP备13007215号-3

      本站所收录作品、热点评论等信息部分来源互联网，目的只是为了系统归纳学习和传递资讯

      所有作品版权归原创作者所有，与本站立场无关，如不慎侵犯了你的权益，请联系我们告知，我们将做删除处理！