trans_handler.go 2.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374
  1. package main
  2. import(
  3. "encoding/json"
  4. "github.com/benbjohnson/go-raft"
  5. "bytes"
  6. "net/http"
  7. "fmt"
  8. "io"
  9. )
  10. type transHandler struct {
  11. name string
  12. client *http.Client
  13. }
  14. // Sends AppendEntries RPCs to a peer when the server is the leader.
  15. func (t transHandler) SendAppendEntriesRequest(server *raft.Server, peer *raft.Peer, req *raft.AppendEntriesRequest) (*raft.AppendEntriesResponse, error) {
  16. var aersp *raft.AppendEntriesResponse
  17. var b bytes.Buffer
  18. json.NewEncoder(&b).Encode(req)
  19. resp, err := Post(&t, fmt.Sprintf("%s/log/append", peer.Name()), &b)
  20. if resp != nil {
  21. defer resp.Body.Close()
  22. aersp = &raft.AppendEntriesResponse{}
  23. if err = json.NewDecoder(resp.Body).Decode(&aersp); err == nil || err == io.EOF {
  24. return aersp, nil
  25. }
  26. }
  27. return aersp, fmt.Errorf("raftd: Unable to append entries: %v", err)
  28. }
  29. // Sends RequestVote RPCs to a peer when the server is the candidate.
  30. func (t transHandler) SendVoteRequest(server *raft.Server, peer *raft.Peer, req *raft.RequestVoteRequest) (*raft.RequestVoteResponse, error) {
  31. var rvrsp *raft.RequestVoteResponse
  32. var b bytes.Buffer
  33. json.NewEncoder(&b).Encode(req)
  34. resp, err := Post(&t, fmt.Sprintf("%s/vote", peer.Name()), &b)
  35. if resp != nil {
  36. defer resp.Body.Close()
  37. rvrsp := &raft.RequestVoteResponse{}
  38. if err = json.NewDecoder(resp.Body).Decode(&rvrsp); err == nil || err == io.EOF {
  39. return rvrsp, nil
  40. }
  41. }
  42. return rvrsp, fmt.Errorf("raftd: Unable to request vote: %v", err)
  43. }
  44. // Sends SnapshotRequest RPCs to a peer when the server is the candidate.
  45. func (t transHandler) SendSnapshotRequest(server *raft.Server, peer *raft.Peer, req *raft.SnapshotRequest) (*raft.SnapshotResponse, error) {
  46. var aersp *raft.SnapshotResponse
  47. var b bytes.Buffer
  48. json.NewEncoder(&b).Encode(req)
  49. debug("[send] POST %s/snapshot [%d %d]", peer.Name(), req.LastTerm, req.LastIndex)
  50. resp, err := Post(&t, fmt.Sprintf("%s/snapshot", peer.Name()), &b)
  51. if resp != nil {
  52. defer resp.Body.Close()
  53. aersp = &raft.SnapshotResponse{}
  54. if err = json.NewDecoder(resp.Body).Decode(&aersp); err == nil || err == io.EOF {
  55. return aersp, nil
  56. }
  57. }
  58. return aersp, fmt.Errorf("raftd: Unable to send snapshot: %v", err)
  59. }