From 6e8bf5dcc16f78fb6a68ac9a401d7d6659d51209 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Tue, 18 Apr 2017 09:43:50 -0500 Subject: [PATCH] move http messenger stuff to separate package --- cluster_test.go | 5 +++-- messenger.go => httpbroadcast/messenger.go | 25 +++++++++++----------- server/server.go | 12 +++++++---- 3 files changed, 24 insertions(+), 18 deletions(-) rename messenger.go => httpbroadcast/messenger.go (85%) diff --git a/cluster_test.go b/cluster_test.go index 1dd6bb486..9dbaa420d 100644 --- a/cluster_test.go +++ b/cluster_test.go @@ -9,6 +9,7 @@ import ( "github.com/davecgh/go-spew/spew" "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/httpbroadcast" ) // Ensure the cluster can fairly distribute partitions across the nodes. @@ -95,10 +96,10 @@ func TestCluster_Health(t *testing.T) { {Host: "serverB:1000"}, {Host: "serverC:1000"}, }, - NodeSet: &pilosa.HTTPNodeSet{}, + NodeSet: &httpbroadcast.HTTPNodeSet{}, } - err := c.NodeSet.(*pilosa.HTTPNodeSet).Join([]*pilosa.Node{ + err := c.NodeSet.(*httpbroadcast.HTTPNodeSet).Join([]*pilosa.Node{ &pilosa.Node{Host: "serverA:1000"}, &pilosa.Node{Host: "serverC:1000"}, &pilosa.Node{Host: "serverD:1000"}, diff --git a/messenger.go b/httpbroadcast/messenger.go similarity index 85% rename from messenger.go rename to httpbroadcast/messenger.go index 37be45c0c..1545e3328 100644 --- a/messenger.go +++ b/httpbroadcast/messenger.go @@ -1,4 +1,4 @@ -package pilosa +package httpbroadcast import ( "bytes" @@ -14,16 +14,17 @@ import ( "net" "github.com/gogo/protobuf/proto" + "github.com/pilosa/pilosa" ) // HTTPBroadcaster represents a NodeSet that broadcasts messages over HTTP. type HTTPBroadcaster struct { - server *Server + server *pilosa.Server internalPort string } // NewHTTPBroadcaster returns a new instance of HTTPBroadcaster. -func NewHTTPBroadcaster(s *Server, internalPort string) *HTTPBroadcaster { +func NewHTTPBroadcaster(s *pilosa.Server, internalPort string) *HTTPBroadcaster { return &HTTPBroadcaster{server: s} } @@ -31,7 +32,7 @@ func NewHTTPBroadcaster(s *Server, internalPort string) *HTTPBroadcaster { // It waits for all nodes to respond before the function returns (and returns any errors). func (h *HTTPBroadcaster) SendSync(pb proto.Message) error { // Marshal the pb to []byte - buf, err := MarshalMessage(pb) + buf, err := pilosa.MarshalMessage(pb) if err != nil { return err } @@ -61,7 +62,7 @@ func (h *HTTPBroadcaster) SendAsync(pb proto.Message) error { return h.SendSync(pb) } -func (h *HTTPBroadcaster) nodes() ([]*Node, error) { +func (h *HTTPBroadcaster) nodes() ([]*pilosa.Node, error) { if h.server == nil { return nil, errors.New("HTTPBroadcaster has no reference to Server.") } @@ -72,7 +73,7 @@ func (h *HTTPBroadcaster) nodes() ([]*Node, error) { return nodeset.Nodes(), nil } -func (h *HTTPBroadcaster) sendNodeMessage(node *Node, msg []byte) error { +func (h *HTTPBroadcaster) sendNodeMessage(node *pilosa.Node, msg []byte) error { var client *http.Client client = http.DefaultClient @@ -114,7 +115,7 @@ func (h *HTTPBroadcaster) sendNodeMessage(node *Node, msg []byte) error { type HTTPBroadcastReceiver struct { port string - handler BroadcastHandler + handler pilosa.BroadcastHandler logOutput io.Writer } @@ -125,7 +126,7 @@ func NewHTTPBroadcastReceiver(port string, logOutput io.Writer) *HTTPBroadcastRe } } -func (rec *HTTPBroadcastReceiver) Start(b BroadcastHandler) error { +func (rec *HTTPBroadcastReceiver) Start(b pilosa.BroadcastHandler) error { rec.handler = b go func() { err := http.ListenAndServe(":"+rec.port, rec) @@ -150,7 +151,7 @@ func (rec *HTTPBroadcastReceiver) ServeHTTP(w http.ResponseWriter, r *http.Reque } // Unmarshal message to specific proto type. - m, err := UnmarshalMessage(body) + m, err := pilosa.UnmarshalMessage(body) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return @@ -164,7 +165,7 @@ func (rec *HTTPBroadcastReceiver) ServeHTTP(w http.ResponseWriter, r *http.Reque // HTTPNodeSet represents a NodeSet that broadcasts messages over HTTP. type HTTPNodeSet struct { - nodes []*Node + nodes []*pilosa.Node } // NewHTTPNodeSet returns a new instance of HTTPNodeSet. @@ -172,7 +173,7 @@ func NewHTTPNodeSet() *HTTPNodeSet { return &HTTPNodeSet{} } -func (h *HTTPNodeSet) Nodes() []*Node { +func (h *HTTPNodeSet) Nodes() []*pilosa.Node { return h.nodes } @@ -180,7 +181,7 @@ func (h *HTTPNodeSet) Open() error { return nil } -func (h *HTTPNodeSet) Join(nodes []*Node) error { +func (h *HTTPNodeSet) Join(nodes []*pilosa.Node) error { h.nodes = nodes return nil } diff --git a/server/server.go b/server/server.go index 80c0457f8..3571dcf3c 100644 --- a/server/server.go +++ b/server/server.go @@ -18,6 +18,7 @@ import ( "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/gossip" + "github.com/pilosa/pilosa/httpbroadcast" ) func init() { @@ -123,10 +124,13 @@ func (m *Command) SetupServer() error { switch m.Config.Cluster.BroadcasterType { // TODO change name to something that encompasses broadcasting, receiving broadcasts, and tracking cluster membership case "http": port := strconv.Itoa(m.Config.Cluster.Gossip.Port) - m.Server.Broadcaster = pilosa.NewHTTPBroadcaster(m.Server, port) - m.Server.BroadcastReceiver = pilosa.NewHTTPBroadcastReceiver(port, m.Stderr) - m.Server.Cluster.NodeSet = pilosa.NewHTTPNodeSet() - m.Server.Cluster.NodeSet.(*pilosa.HTTPNodeSet).Join(m.Server.Cluster.Nodes) + m.Server.Broadcaster = httpbroadcast.NewHTTPBroadcaster(m.Server, port) + m.Server.BroadcastReceiver = httpbroadcast.NewHTTPBroadcastReceiver(port, m.Stderr) + m.Server.Cluster.NodeSet = httpbroadcast.NewHTTPNodeSet() + err := m.Server.Cluster.NodeSet.(*httpbroadcast.HTTPNodeSet).Join(m.Server.Cluster.Nodes) + if err != nil { + return err + } case "gossip": gossipPort, err := strconv.Atoi(pilosa.DefaultGossipPort) if err != nil {