scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

move http messenger stuff to separate package

Browse source
This commit is contained in:
Matt Jaffee 2017-04-18 09:43:50 -05:00 • committed by Travis
parent d2621709f1
commit 6e8bf5dcc1
3 changed files with 24 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
cluster_test.go
View file

@ -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"},

25
messenger.go → httpbroadcast/messenger.go
View file

@ -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
}

12
server/server.go
View file

@ -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 {