(cfg *Config)
| 63 | const workerCount int = 2 |
| 64 | |
| 65 | func newConn(cfg *Config) (c *conn, err error) { |
| 66 | // get local node id |
| 67 | var localNodeID proto.NodeID |
| 68 | if localNodeID, err = kms.GetLocalNodeID(); err != nil { |
| 69 | return |
| 70 | } |
| 71 | |
| 72 | // get local private key |
| 73 | var privKey *asymmetric.PrivateKey |
| 74 | if privKey, err = kms.GetLocalPrivateKey(); err != nil { |
| 75 | return |
| 76 | } |
| 77 | |
| 78 | c = &conn{ |
| 79 | dbID: proto.DatabaseID(cfg.DatabaseID), |
| 80 | localNodeID: localNodeID, |
| 81 | privKey: privKey, |
| 82 | queries: make([]types.Query, 0), |
| 83 | } |
| 84 | |
| 85 | // get peers from BP |
| 86 | var peers *proto.Peers |
| 87 | if peers, err = cacheGetPeers(c.dbID, c.privKey); err != nil { |
| 88 | return nil, errors.WithMessage(err, "cacheGetPeers failed") |
| 89 | } |
| 90 | |
| 91 | if cfg.Mirror != "" { |
| 92 | c.leader = &pconn{ |
| 93 | wg: &sync.WaitGroup{}, |
| 94 | parent: c, |
| 95 | pCaller: mux.NewRawCaller(cfg.Mirror), |
| 96 | } |
| 97 | |
| 98 | // no ack workers required, mirror mode does not support ack worker |
| 99 | } else { |
| 100 | if cfg.UseLeader { |
| 101 | var caller rpc.PCaller |
| 102 | if cfg.UseDirectRPC { |
| 103 | caller = rpc.NewPersistentCaller(peers.Leader) |
| 104 | } else { |
| 105 | caller = mux.NewPersistentCaller(peers.Leader) |
| 106 | } |
| 107 | c.leader = &pconn{ |
| 108 | wg: &sync.WaitGroup{}, |
| 109 | ackCh: make(chan *types.Ack, workerCount*4), |
| 110 | parent: c, |
| 111 | pCaller: caller, |
| 112 | } |
| 113 | } |
| 114 | |
| 115 | // choose a random follower node |
| 116 | if cfg.UseFollower && len(peers.Servers) > 1 { |
| 117 | for { |
| 118 | node := peers.Servers[randSource.Intn(len(peers.Servers))] |
| 119 | if node != peers.Leader { |
| 120 | var caller rpc.PCaller |
| 121 | if cfg.UseDirectRPC { |
| 122 | caller = rpc.NewPersistentCaller(node) |
no test coverage detected