|
@@ -87,7 +87,7 @@ func (t *transporter) SendAppendEntriesRequest(server raft.Server, peer *raft.Pe
|
|
|
|
|
|
|
|
start := time.Now()
|
|
start := time.Now()
|
|
|
|
|
|
|
|
- resp, httpRequest, err := t.Post(fmt.Sprintf("%s/log/append", u), &b)
|
|
|
|
|
|
|
+ resp, _, err := t.Post(fmt.Sprintf("%s/log/append", u), &b)
|
|
|
|
|
|
|
|
end := time.Now()
|
|
end := time.Now()
|
|
|
|
|
|
|
@@ -106,8 +106,6 @@ func (t *transporter) SendAppendEntriesRequest(server raft.Server, peer *raft.Pe
|
|
|
if resp != nil {
|
|
if resp != nil {
|
|
|
defer resp.Body.Close()
|
|
defer resp.Body.Close()
|
|
|
|
|
|
|
|
- t.CancelWhenTimeout(httpRequest)
|
|
|
|
|
-
|
|
|
|
|
aeresp := &raft.AppendEntriesResponse{}
|
|
aeresp := &raft.AppendEntriesResponse{}
|
|
|
if _, err = aeresp.Decode(resp.Body); err != nil && err != io.EOF {
|
|
if _, err = aeresp.Decode(resp.Body); err != nil && err != io.EOF {
|
|
|
log.Warn("transporter.ae.decoding.error:", err)
|
|
log.Warn("transporter.ae.decoding.error:", err)
|
|
@@ -131,7 +129,7 @@ func (t *transporter) SendVoteRequest(server raft.Server, peer *raft.Peer, req *
|
|
|
u, _ := t.registry.PeerURL(peer.Name)
|
|
u, _ := t.registry.PeerURL(peer.Name)
|
|
|
log.Debugf("Send Vote from %s to %s", server.Name(), u)
|
|
log.Debugf("Send Vote from %s to %s", server.Name(), u)
|
|
|
|
|
|
|
|
- resp, httpRequest, err := t.Post(fmt.Sprintf("%s/vote", u), &b)
|
|
|
|
|
|
|
+ resp, _, err := t.Post(fmt.Sprintf("%s/vote", u), &b)
|
|
|
|
|
|
|
|
if err != nil {
|
|
if err != nil {
|
|
|
log.Debugf("Cannot send VoteRequest to %s : %s", u, err)
|
|
log.Debugf("Cannot send VoteRequest to %s : %s", u, err)
|
|
@@ -140,8 +138,6 @@ func (t *transporter) SendVoteRequest(server raft.Server, peer *raft.Peer, req *
|
|
|
if resp != nil {
|
|
if resp != nil {
|
|
|
defer resp.Body.Close()
|
|
defer resp.Body.Close()
|
|
|
|
|
|
|
|
- t.CancelWhenTimeout(httpRequest)
|
|
|
|
|
-
|
|
|
|
|
rvrsp := &raft.RequestVoteResponse{}
|
|
rvrsp := &raft.RequestVoteResponse{}
|
|
|
if _, err = rvrsp.Decode(resp.Body); err != nil && err != io.EOF {
|
|
if _, err = rvrsp.Decode(resp.Body); err != nil && err != io.EOF {
|
|
|
log.Warn("transporter.vr.decoding.error:", err)
|
|
log.Warn("transporter.vr.decoding.error:", err)
|
|
@@ -164,7 +160,7 @@ func (t *transporter) SendSnapshotRequest(server raft.Server, peer *raft.Peer, r
|
|
|
u, _ := t.registry.PeerURL(peer.Name)
|
|
u, _ := t.registry.PeerURL(peer.Name)
|
|
|
log.Debugf("Send Snapshot Request from %s to %s", server.Name(), u)
|
|
log.Debugf("Send Snapshot Request from %s to %s", server.Name(), u)
|
|
|
|
|
|
|
|
- resp, httpRequest, err := t.Post(fmt.Sprintf("%s/snapshot", u), &b)
|
|
|
|
|
|
|
+ resp, _, err := t.Post(fmt.Sprintf("%s/snapshot", u), &b)
|
|
|
|
|
|
|
|
if err != nil {
|
|
if err != nil {
|
|
|
log.Debugf("Cannot send Snapshot Request to %s : %s", u, err)
|
|
log.Debugf("Cannot send Snapshot Request to %s : %s", u, err)
|
|
@@ -173,8 +169,6 @@ func (t *transporter) SendSnapshotRequest(server raft.Server, peer *raft.Peer, r
|
|
|
if resp != nil {
|
|
if resp != nil {
|
|
|
defer resp.Body.Close()
|
|
defer resp.Body.Close()
|
|
|
|
|
|
|
|
- t.CancelWhenTimeout(httpRequest)
|
|
|
|
|
-
|
|
|
|
|
ssrsp := &raft.SnapshotResponse{}
|
|
ssrsp := &raft.SnapshotResponse{}
|
|
|
if _, err = ssrsp.Decode(resp.Body); err != nil && err != io.EOF {
|
|
if _, err = ssrsp.Decode(resp.Body); err != nil && err != io.EOF {
|
|
|
log.Warn("transporter.ss.decoding.error:", err)
|
|
log.Warn("transporter.ss.decoding.error:", err)
|
|
@@ -197,7 +191,7 @@ func (t *transporter) SendSnapshotRecoveryRequest(server raft.Server, peer *raft
|
|
|
u, _ := t.registry.PeerURL(peer.Name)
|
|
u, _ := t.registry.PeerURL(peer.Name)
|
|
|
log.Debugf("Send Snapshot Recovery from %s to %s", server.Name(), u)
|
|
log.Debugf("Send Snapshot Recovery from %s to %s", server.Name(), u)
|
|
|
|
|
|
|
|
- resp, httpRequest, err := t.Post(fmt.Sprintf("%s/snapshotRecovery", u), &b)
|
|
|
|
|
|
|
+ resp, _, err := t.Post(fmt.Sprintf("%s/snapshotRecovery", u), &b)
|
|
|
|
|
|
|
|
if err != nil {
|
|
if err != nil {
|
|
|
log.Debugf("Cannot send Snapshot Recovery to %s : %s", u, err)
|
|
log.Debugf("Cannot send Snapshot Recovery to %s : %s", u, err)
|
|
@@ -206,8 +200,6 @@ func (t *transporter) SendSnapshotRecoveryRequest(server raft.Server, peer *raft
|
|
|
if resp != nil {
|
|
if resp != nil {
|
|
|
defer resp.Body.Close()
|
|
defer resp.Body.Close()
|
|
|
|
|
|
|
|
- t.CancelWhenTimeout(httpRequest)
|
|
|
|
|
-
|
|
|
|
|
ssrrsp := &raft.SnapshotRecoveryResponse{}
|
|
ssrrsp := &raft.SnapshotRecoveryResponse{}
|
|
|
if _, err = ssrrsp.Decode(resp.Body); err != nil && err != io.EOF {
|
|
if _, err = ssrrsp.Decode(resp.Body); err != nil && err != io.EOF {
|
|
|
log.Warn("transporter.ssr.decoding.error:", err)
|
|
log.Warn("transporter.ssr.decoding.error:", err)
|
|
@@ -232,11 +224,3 @@ func (t *transporter) Get(urlStr string) (*http.Response, *http.Request, error)
|
|
|
resp, err := t.client.Do(req)
|
|
resp, err := t.client.Do(req)
|
|
|
return resp, req, err
|
|
return resp, req, err
|
|
|
}
|
|
}
|
|
|
-
|
|
|
|
|
-// Cancel the on fly HTTP transaction when timeout happens.
|
|
|
|
|
-func (t *transporter) CancelWhenTimeout(req *http.Request) {
|
|
|
|
|
- go func() {
|
|
|
|
|
- time.Sleep(t.requestTimeout)
|
|
|
|
|
- t.transport.CancelRequest(req)
|
|
|
|
|
- }()
|
|
|
|
|
-}
|
|
|