wip implement more... still quite broken

api.go

@@ -517,7 +517,7 @@ func (api *API) ClusterMessage(ctx context.Context, reqBody io.Reader) error {
 	}
 
 	// Forward the error message.
-	if err := api.server.receiveMessage(pb); err != nil {
+	if err := api.server.receiveMessage(decode(pb)); err != nil {
 		return errors.Wrap(err, "receiving message")
 	}
 	return nil

broadcast.go

@@ -73,8 +73,7 @@ const (
 )
 
 // MarshalMessage encodes the protobuf message into a byte slice.
-func MarshalMessage(pm Message) ([]byte, error) {
-	m := encode(pm)
+func MarshalMessage(m proto.Message) ([]byte, error) {
 	var typ uint8
 	switch obj := m.(type) {
 	case *internal.CreateShardMessage:
@@ -120,19 +119,45 @@ func MarshalMessage(pm Message) ([]byte, error) {
 }
 
 func encode(m Message) proto.Message {
-	var pm proto.Message
 	switch mt := m.(type) {
 	case *CreateShardMessage:
 		return encodeCreateShardMessage(mt)
 	case *CreateIndexMessage:
 		return encodeCreateIndexMessage(mt)
-		// TODO, the rest
+	case *DeleteIndexMessage:
+		return encodeDeleteIndexMessage(mt)
+	case *CreateFieldMessage:
+		return encodeCreateFieldMessage(mt)
+	case *DeleteFieldMessage:
+		return encodeDeleteFieldMessage(mt)
+	case *CreateViewMessage:
+		return encodeCreateViewMessage(mt)
+	case *DeleteViewMessage:
+		return encodeDeleteViewMessage(mt)
+	case *ClusterStatus:
+		return encodeClusterStatus(mt)
+	case *ResizeInstruction:
+		return encodeResizeInstruction(mt)
+	case *ResizeInstructionComplete:
+		return encodeResizeInstructionComplete(mt)
+	case *SetCoordinatorMessage:
+		return encodeSetCoordinatorMessage(mt)
+	case *UpdateCoordinatorMessage:
+		return encodeUpdateCoordinatorMessage(mt)
+	case *NodeStateMessage:
+		return encodeNodeStateMessage(mt)
+	case *RecalculateCaches:
+		return encodeRecalculateCaches(mt)
+	case *nodeEvent:
+		return encodeNodeEventMessage(mt)
+	case *NodeStatus:
+		return encodeNodeStatus(mt)
 	}
 	return nil
 }
 
 // UnmarshalMessage decodes the byte slice into a protobuf message.
-func UnmarshalMessage(buf []byte) (Message, error) {
+func UnmarshalMessage(buf []byte) (proto.Message, error) {
 	typ, buf := buf[0], buf[1:]
 	var m proto.Message
 	switch typ {
@@ -177,3 +202,41 @@ func UnmarshalMessage(buf []byte) (Message, error) {
 	}
 	return m, nil
 }
+
+func decode(m proto.Message) Message {
+	switch mt := m.(type) {
+	case *internal.CreateShardMessage:
+		return decodeCreateShardMessage(mt)
+	case *internal.CreateIndexMessage:
+		return decodeCreateIndexMessage(mt)
+	case *internal.DeleteIndexMessage:
+		return decodeDeleteIndexMessage(mt)
+	case *internal.CreateFieldMessage:
+		return decodeCreateFieldMessage(mt)
+	case *internal.DeleteFieldMessage:
+		return decodeDeleteFieldMessage(mt)
+	case *internal.CreateViewMessage:
+		return decodeCreateViewMessage(mt)
+	case *internal.DeleteViewMessage:
+		return decodeDeleteViewMessage(mt)
+	case *internal.ClusterStatus:
+		return decodeClusterStatus(mt)
+	case *internal.ResizeInstruction:
+		return decodeResizeInstruction(mt)
+	case *internal.ResizeInstructionComplete:
+		return decodeResizeInstructionComplete(mt)
+	case *internal.SetCoordinatorMessage:
+		return decodeSetCoordinatorMessage(mt)
+	case *internal.UpdateCoordinatorMessage:
+		return decodeUpdateCoordinatorMessage(mt)
+	case *internal.NodeStateMessage:
+		return decodeNodeStateMessage(mt)
+	case *internal.RecalculateCaches:
+		return decodeRecalculateCaches(mt)
+	case *internal.NodeEventMessage:
+		return decodeNodeEventMessage(mt)
+	case *internal.NodeStatus:
+		return decodeNodeStatus(mt)
+	}
+	return nil
+}

cluster.go

@@ -268,8 +268,8 @@ func (c *cluster) setCoordinator(n *Node) error {
 	c.mu.Unlock()
 	// Send the update coordinator message to all nodes.
 	err := c.broadcaster.SendSync(
-		&internal.UpdateCoordinatorMessage{
-			New: EncodeNode(n),
+		&UpdateCoordinatorMessage{
+			New: n,
 		})
 	if err != nil {
 		return fmt.Errorf("problem sending UpdateCoordinator message: %v", err)
@@ -423,7 +423,7 @@ func (c *cluster) setNodeState(state string) error {
 	}
 
 	// Send node state to coordinator.
-	ns := &internal.NodeStateMessage{
+	ns := &NodeStateMessage{
 		NodeID: c.Node.ID,
 		State:  state,
 	}
@@ -460,12 +460,12 @@ func (c *cluster) receiveNodeState(nodeID string, state string) error {
 	return nil
 }
 
-// Status returns the internal ClusterStatus representation.
-func (c *cluster) Status() *internal.ClusterStatus {
-	return &internal.ClusterStatus{
+// Status returns the the cluster's status including what nodes it contains, it's ID, and current state.
+func (c *cluster) Status() *ClusterStatus {
+	return &ClusterStatus{
 		ClusterID: c.id,
 		State:     c.state,
-		Nodes:     EncodeNodes(c.Nodes),
+		Nodes:     c.Nodes,
 	}
 }
 
@@ -640,8 +640,8 @@ func (c *cluster) diff(other *cluster) (action string, nodeID string, err error)
 
 // fragSources returns a list of ResizeSources - for each node in the `to` cluster -
 // required to move from cluster `c` to cluster `to`.
-func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*internal.ResizeSource, error) {
-	m := make(map[string][]*internal.ResizeSource)
+func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*ResizeSource, error) {
+	m := make(map[string][]*ResizeSource)
 
 	// Determine if a node is being added or removed.
 	action, diffNodeID, err := c.diff(to)
@@ -700,7 +700,7 @@ func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*internal.R
 
 	// Get the ResizeSource for each diff.
 	for nodeID, diff := range diffs {
-		m[nodeID] = []*internal.ResizeSource{}
+		m[nodeID] = []*ResizeSource{}
 		for _, frag := range diff {
 			// If there is no valid source node ID for a fragment,
 			// it likely means that the replica factor was not
@@ -711,8 +711,8 @@ func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*internal.R
 				return nil, errors.New("not enough data to perform resize (replica factor may need to be increased)")
 			}
 
-			src := &internal.ResizeSource{
-				Node:  EncodeNode(c.unprotectedNodeByID(srcNodeID)),
+			src := &ResizeSource{
+				Node:  c.unprotectedNodeByID(srcNodeID),
 				Index: idx.Name(),
 				Field: frag.field,
 				View:  frag.view,
@@ -856,9 +856,9 @@ func (c *cluster) waitForStarted() error {
 		// TODO: Because the normal code path already sends a NodeJoin event (via
 		// memberlist), this it a bit redundant in most cases. Perhaps determine
 		// that the node has been restarted and don't do this step.
-		msg := &internal.NodeEventMessage{
-			Event: uint32(NodeJoin),
-			Node:  EncodeNode(c.Node),
+		msg := &nodeEvent{
+			Event: NodeJoin,
+			Node:  c.Node,
 		}
 		if err := c.broadcaster.SendSync(msg); err != nil {
 			return fmt.Errorf("sending restart NodeJoin: %v", err)
@@ -968,8 +968,8 @@ func (c *cluster) setStateAndBroadcast(state string) error {
 	return c.broadcaster.SendSync(c.Status())
 }
 
-func (c *cluster) sendTo(node *Node, msg proto.Message) error {
-	if err := c.broadcaster.SendTo(node, msg); err != nil {
+func (c *cluster) sendTo(node *Node, m Message) error {
+	if err := c.broadcaster.SendTo(node, m); err != nil {
 		return errors.Wrap(err, "sending")
 	}
 	return nil
@@ -1075,7 +1075,7 @@ func (c *cluster) generateResizeJobByAction(nodeAction nodeAction) (*resizeJob,
 	}
 
 	// multiIndex is a map of sources initialized with all the nodes in toCluster.
-	multiIndex := make(map[string][]*internal.ResizeSource)
+	multiIndex := make(map[string][]*ResizeSource)
 
 	for _, n := range toCluster.Nodes {
 		multiIndex[n.ID] = nil
@@ -1099,12 +1099,12 @@ func (c *cluster) generateResizeJobByAction(nodeAction nodeAction) (*resizeJob,
 			j.IDs[id] = true
 			continue
 		}
-		instr := &internal.ResizeInstruction{
+		instr := &ResizeInstruction{
 			JobID:         j.ID,
-			Node:          EncodeNode(toCluster.unprotectedNodeByID(id)),
-			Coordinator:   EncodeNode(c.coordinatorNode()),
+			Node:          toCluster.unprotectedNodeByID(id),
+			Coordinator:   c.coordinatorNode(),
 			Sources:       sources,
-			Schema:        c.holder.encodeSchema(), // Include the schema to ensure it's in sync on the receiving node.
+			Schema:        &Schema{Indexes: c.holder.Schema()}, // Include the schema to ensure it's in sync on the receiving node.
 			ClusterStatus: c.Status(),
 		}
 		j.Instructions = append(j.Instructions, instr)
@@ -1148,7 +1148,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
 		<-c.holder.opened
 
 		// Prepare the return message.
-		complete := &internal.ResizeInstructionComplete{
+		complete := &ResizeInstructionComplete{
 			JobID: instr.JobID,
 			Node:  instr.Node,
 			Error: "",
@@ -1167,7 +1167,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
 			for _, src := range instr.Sources {
 				c.logger.Printf("get shard %d for index %s from host %s", src.Shard, src.Index, src.Node.URI)
 
-				srcURI := decodeURI(src.Node.URI)
+				srcURI := src.Node.URI
 
 				// Retrieve field.
 				f := c.holder.Field(src.Index, src.Field)
@@ -1219,14 +1219,14 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
 			complete.Error = err.Error()
 		}
 
-		if err := c.sendTo(DecodeNode(instr.Coordinator), complete); err != nil {
+		if err := c.sendTo(instr.Coordinator, complete); err != nil {
 			c.logger.Printf("sending resizeInstructionComplete error: err=%s", err)
 		}
 	}()
 	return nil
 }
 
-func (c *cluster) markResizeInstructionComplete(complete *internal.ResizeInstructionComplete) error {
+func (c *cluster) markResizeInstructionComplete(complete *ResizeInstructionComplete) error {
 
 	j := c.job(complete.JobID)
 
@@ -1263,7 +1263,7 @@ func (c *cluster) job(id int64) *resizeJob {
 type resizeJob struct {
 	ID           int64
 	IDs          map[string]bool
-	Instructions []*internal.ResizeInstruction
+	Instructions []*ResizeInstruction
 	Broadcaster  broadcaster
 
 	action string
@@ -1366,7 +1366,7 @@ func (j *resizeJob) distributeResizeInstructions() error {
 		// a dummy node object to use in the SendTo() method.
 		node := &Node{
 			ID:  instr.Node.ID,
-			URI: decodeURI(instr.Node.URI),
+			URI: instr.Node.URI,
 		}
 		j.Logger.Printf("send resize instructions: %v", instr)
 		if err := j.Broadcaster.SendTo(node, instr); err != nil {
@@ -1757,7 +1757,7 @@ type ResizeInstruction struct {
 	ClusterStatus *ClusterStatus
 }
 
-func decodeResizeInstruction(ri *internal.ResizeInstruction) ResizeInstruction {
+func decodeResizeInstruction(ri *internal.ResizeInstruction) *ResizeInstruction {
 	return &ResizeInstruction{
 		JobID:         ri.JobID,
 		Node:          DecodeNode(ri.Node),
@@ -1768,6 +1768,17 @@ func decodeResizeInstruction(ri *internal.ResizeInstruction) ResizeInstruction {
 	}
 }
 
+func encodeResizeInstruction(m *ResizeInstruction) *internal.ResizeInstruction {
+	return &internal.ResizeInstruction{
+		JobID:         m.JobID,
+		Node:          EncodeNode(m.Node),
+		Coordinator:   EncodeNode(m.Coordinator),
+		Sources:       encodeResizeSources(m.Sources),
+		Schema:        encodeSchema(m.Schema),
+		ClusterStatus: encodeClusterStatus(m.ClusterStatus),
+	}
+}
+
 type ResizeSource struct {
 	Node  *Node  `protobuf:"bytes,1,opt,name=Node" json:"Node,omitempty"`
 	Index string `protobuf:"bytes,2,opt,name=Index,proto3" json:"Index,omitempty"`
@@ -1784,6 +1795,14 @@ func decodeResizeSources(srcs []*internal.ResizeSource) []*ResizeSource {
 	return new
 }
 
+func encodeResizeSources(srcs []*ResizeSource) []*internal.ResizeSource {
+	new := make([]*internal.ResizeSource, 0, len(srcs))
+	for _, src := range srcs {
+		new = append(new, encodeResizeSource(src))
+	}
+	return new
+}
+
 func decodeResizeSource(rs *internal.ResizeSource) *ResizeSource {
 	return &ResizeSource{
 		Node:  DecodeNode(rs.Node),
@@ -1794,6 +1813,16 @@ func decodeResizeSource(rs *internal.ResizeSource) *ResizeSource {
 	}
 }
 
+func encodeResizeSource(m *ResizeSource) *internal.ResizeSource {
+	return &internal.ResizeSource{
+		Node:  EncodeNode(m.Node),
+		Index: m.Index,
+		Field: m.Field,
+		View:  m.View,
+		Shard: m.Shard,
+	}
+}
+
 // Schema is a schema
 type Schema struct {
 	Indexes []*IndexInfo
@@ -1805,6 +1834,12 @@ func decodeSchema(s *internal.Schema) *Schema {
 	}
 }
 
+func encodeSchema(m *Schema) *internal.Schema {
+	return &internal.Schema{
+		Indexes: encodeIndexInfos(m.Indexes),
+	}
+}
+
 func decodeIndexes(idxs []*internal.Index) []*IndexInfo {
 	new := make([]*IndexInfo, 0, len(idxs))
 	for _, idx := range idxs {
@@ -1813,6 +1848,14 @@ func decodeIndexes(idxs []*internal.Index) []*IndexInfo {
 	return new
 }
 
+func encodeIndexInfos(idxs []*IndexInfo) []*internal.Index {
+	new := make([]*internal.Index, 0, len(idxs))
+	for _, idx := range idxs {
+		new = append(new, encodeIndexInfo(idx))
+	}
+	return new
+}
+
 func decodeIndex(idx *internal.Index) *IndexInfo {
 	return &IndexInfo{
 		Name:   idx.Name,
@@ -1820,6 +1863,13 @@ func decodeIndex(idx *internal.Index) *IndexInfo {
 	}
 }
 
+func encodeIndexInfo(idx *IndexInfo) *internal.Index {
+	return &internal.Index{
+		Name:   idx.Name,
+		Fields: encodeFieldInfos(idx.Fields),
+	}
+}
+
 func decodeFields(fs []*internal.Field) []*FieldInfo {
 	new := make([]*FieldInfo, 0, len(fs))
 	for _, f := range fs {
@@ -1828,6 +1878,14 @@ func decodeFields(fs []*internal.Field) []*FieldInfo {
 	return new
 }
 
+func encodeFieldInfos(fs []*FieldInfo) []*internal.Field {
+	new := make([]*internal.Field, 0, len(fs))
+	for _, f := range fs {
+		new = append(new, encodeFieldInfo(f))
+	}
+	return new
+}
+
 func decodeField(f *internal.Field) *FieldInfo {
 	fi := &FieldInfo{
 		Name:    f.Name,
@@ -1840,6 +1898,19 @@ func decodeField(f *internal.Field) *FieldInfo {
 	return fi
 }
 
+func encodeFieldInfo(f *FieldInfo) *internal.Field {
+	ifield := &internal.Field{
+		Name:  f.Name,
+		Meta:  encodeFieldOptions(&f.Options),
+		Views: make([]string, 0, len(f.Views)),
+	}
+
+	for _, viewinfo := range f.Views {
+		ifield.Views = append(ifield.Views, viewinfo.Name)
+	}
+	return ifield
+}
+
 // EncodeNodes converts a slice of Nodes into its internal representation.
 func EncodeNodes(a []*Node) []*internal.Node {
 	other := make([]*internal.Node, len(a))
@@ -1878,6 +1949,14 @@ func decodeClusterStatus(cs *internal.ClusterStatus) *ClusterStatus {
 	}
 }
 
+func encodeClusterStatus(m *ClusterStatus) *internal.ClusterStatus {
+	return &internal.ClusterStatus{
+		State:     m.State,
+		ClusterID: m.ClusterID,
+		Nodes:     EncodeNodes(m.Nodes),
+	}
+}
+
 // DecodeNode converts a proto message into a Node.
 func DecodeNode(node *internal.Node) *Node {
 	return &Node{
@@ -1969,3 +2048,223 @@ func decodeIndexMeta(pb *internal.IndexMeta) *IndexOptions {
 		Keys: pb.Keys,
 	}
 }
+
+type DeleteIndexMessage struct {
+	Index string
+}
+
+func encodeDeleteIndexMessage(m *DeleteIndexMessage) *internal.DeleteIndexMessage {
+	return &internal.DeleteIndexMessage{
+		Index: m.Index,
+	}
+}
+
+func decodeDeleteIndexMessage(pb *internal.DeleteIndexMessage) *DeleteIndexMessage {
+	return &DeleteIndexMessage{
+		Index: pb.Index,
+	}
+}
+
+type CreateFieldMessage struct {
+	Index string
+	Field string
+	Meta  *FieldOptions
+}
+
+func encodeCreateFieldMessage(m *CreateFieldMessage) *internal.CreateFieldMessage {
+	return &internal.CreateFieldMessage{
+		Index: m.Index,
+		Field: m.Field,
+		Meta:  encodeFieldOptions(m.Meta),
+	}
+}
+
+func decodeCreateFieldMessage(pb *internal.CreateFieldMessage) *CreateFieldMessage {
+	return &CreateFieldMessage{
+		Index: pb.Index,
+		Field: pb.Field,
+		Meta:  decodeFieldOptions(pb.Meta),
+	}
+}
+
+type DeleteFieldMessage struct {
+	Index string
+	Field string
+}
+
+func encodeDeleteFieldMessage(m *DeleteFieldMessage) *internal.DeleteFieldMessage {
+	return &internal.DeleteFieldMessage{
+		Index: m.Index,
+		Field: m.Field,
+	}
+}
+
+func decodeDeleteFieldMessage(pb *internal.DeleteFieldMessage) *DeleteFieldMessage {
+	return &DeleteFieldMessage{
+		Index: pb.Index,
+		Field: pb.Field,
+	}
+}
+
+type CreateViewMessage struct {
+	Index string
+	Field string
+	View  string
+}
+
+func encodeCreateViewMessage(m *CreateViewMessage) *internal.CreateViewMessage {
+	return &internal.CreateViewMessage{
+		Index: m.Index,
+		Field: m.Field,
+		View:  m.View,
+	}
+}
+
+func decodeCreateViewMessage(pb *internal.CreateViewMessage) *CreateViewMessage {
+	return &CreateViewMessage{
+		Index: pb.Index,
+		Field: pb.Field,
+		View:  pb.View,
+	}
+}
+
+type DeleteViewMessage struct {
+	Index string
+	Field string
+	View  string
+}
+
+func encodeDeleteViewMessage(m *DeleteViewMessage) *internal.DeleteViewMessage {
+	return &internal.DeleteViewMessage{
+		Index: m.Index,
+		Field: m.Field,
+		View:  m.View,
+	}
+}
+
+func decodeDeleteViewMessage(pb *internal.DeleteViewMessage) *DeleteViewMessage {
+	return &DeleteViewMessage{
+		Index: pb.Index,
+		Field: pb.Field,
+		View:  pb.View,
+	}
+}
+
+type ResizeInstructionComplete struct {
+	JobID int64
+	Node  *Node
+	Error string
+}
+
+func encodeResizeInstructionComplete(m *ResizeInstructionComplete) *internal.ResizeInstructionComplete {
+	return &internal.ResizeInstructionComplete{
+		JobID: m.JobID,
+		Node:  EncodeNode(m.Node),
+		Error: m.Error,
+	}
+}
+
+func decodeResizeInstructionComplete(pb *internal.ResizeInstructionComplete) *ResizeInstructionComplete {
+	return &ResizeInstructionComplete{
+		JobID: pb.JobID,
+		Node:  DecodeNode(pb.Node),
+		Error: pb.Error,
+	}
+}
+
+type SetCoordinatorMessage struct {
+	New *Node
+}
+
+func encodeSetCoordinatorMessage(m *SetCoordinatorMessage) *internal.SetCoordinatorMessage {
+	return &internal.SetCoordinatorMessage{
+		New: EncodeNode(m.New),
+	}
+}
+
+func decodeSetCoordinatorMessage(pb *internal.SetCoordinatorMessage) *SetCoordinatorMessage {
+	return &SetCoordinatorMessage{
+		New: DecodeNode(pb.New),
+	}
+}
+
+type UpdateCoordinatorMessage struct {
+	New *Node
+}
+
+func encodeUpdateCoordinatorMessage(m *UpdateCoordinatorMessage) *internal.UpdateCoordinatorMessage {
+	return &internal.UpdateCoordinatorMessage{
+		New: EncodeNode(m.New),
+	}
+}
+
+func decodeUpdateCoordinatorMessage(pb *internal.UpdateCoordinatorMessage) *UpdateCoordinatorMessage {
+	return &UpdateCoordinatorMessage{
+		New: DecodeNode(pb.New),
+	}
+}
+
+type NodeStateMessage struct {
+	NodeID string `protobuf:"bytes,1,opt,name=NodeID,proto3" json:"NodeID,omitempty"`
+	State  string `protobuf:"bytes,2,opt,name=State,proto3" json:"State,omitempty"`
+}
+
+func encodeNodeStateMessage(m *NodeStateMessage) *internal.NodeStateMessage {
+	return &internal.NodeStateMessage{
+		NodeID: m.NodeID,
+		State:  m.State,
+	}
+}
+
+func decodeNodeStateMessage(pb *internal.NodeStateMessage) *NodeStateMessage {
+	return &NodeStateMessage{
+		NodeID: pb.NodeID,
+		State:  pb.State,
+	}
+}
+
+func encodeNodeEventMessage(m *nodeEvent) *internal.NodeEventMessage {
+	return &internal.NodeEventMessage{
+		Event: uint32(m.Event),
+		Node:  EncodeNode(m.Node),
+	}
+}
+
+func decodeNodeEventMessage(pb *internal.NodeEventMessage) *nodeEvent {
+	return &nodeEvent{
+		Event: NodeEventType(pb.Event),
+		Node:  DecodeNode(pb.Node),
+	}
+}
+
+type NodeStatus struct {
+	Node      *Node
+	MaxShards map[string]uint64
+	Schema    *Schema
+}
+
+func encodeNodeStatus(m *NodeStatus) *internal.NodeStatus {
+	return &internal.NodeStatus{
+		Node:      EncodeNode(m.Node),
+		MaxShards: &internal.MaxShards{Standard: m.MaxShards},
+		Schema:    encodeSchema(m.Schema),
+	}
+}
+
+func decodeNodeStatus(pb *internal.NodeStatus) *NodeStatus {
+	return &NodeStatus{
+		Node:      DecodeNode(pb.Node),
+		MaxShards: pb.MaxShards.Standard,
+		Schema:    decodeSchema(pb.Schema),
+	}
+}
+
+type RecalculateCaches struct{}
+
+func decodeRecalculateCaches(pb *internal.RecalculateCaches) *RecalculateCaches {
+	return &RecalculateCaches{}
+}
+
+func encodeRecalculateCaches(*RecalculateCaches) *internal.RecalculateCaches {
+	return &internal.RecalculateCaches{}
+}

holder.go

@@ -230,7 +230,7 @@ func (h *Holder) Schema() []*IndexInfo {
 }
 
 // applySchema applies an internal Schema to Holder.
-func (h *Holder) applySchema(schema *internal.Schema) error {
+func (h *Holder) applySchema(schema *Schema) error {
 	// Create indexes that don't exist.
 	for _, index := range schema.Indexes {
 		opt := IndexOptions{}
@@ -240,14 +240,13 @@ func (h *Holder) applySchema(schema *internal.Schema) error {
 		}
 		// Create fields that don't exist.
 		for _, f := range index.Fields {
-			opt := decodeFieldOptions(f.Meta)
-			field, err := idx.CreateFieldIfNotExists(f.Name, *opt)
+			field, err := idx.CreateFieldIfNotExists(f.Name, f.Options)
 			if err != nil {
 				return errors.Wrap(err, "creating field")
 			}
 			// Create views that don't exist.
 			for _, v := range f.Views {
-				_, err := field.createViewIfNotExists(v)
+				_, err := field.createViewIfNotExists(v.Name)
 				if err != nil {
 					return errors.Wrap(err, "creating view")
 				}

server.go

@@ -27,8 +27,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/gogo/protobuf/proto"
-	"github.com/pilosa/pilosa/internal"
 	"github.com/pkg/errors"
 
 	"golang.org/x/sync/errgroup"
@@ -431,40 +429,40 @@ func (s *Server) monitorAntiEntropy() {
 }
 
 // receiveMessage represents an implementation of BroadcastHandler.
-func (s *Server) receiveMessage(pb proto.Message) error {
-	switch obj := pb.(type) {
-	case *internal.CreateShardMessage:
+func (s *Server) receiveMessage(m Message) error {
+	switch obj := m.(type) {
+	case *CreateShardMessage:
 		idx := s.holder.Index(obj.Index)
 		if idx == nil {
 			return fmt.Errorf("Local Index not found: %s", obj.Index)
 		}
 		idx.setRemoteMaxShard(obj.Shard)
-	case *internal.CreateIndexMessage:
+	case *CreateIndexMessage:
 		opt := IndexOptions{}
 		_, err := s.holder.CreateIndex(obj.Index, opt)
 		if err != nil {
 			return err
 		}
-	case *internal.DeleteIndexMessage:
+	case *DeleteIndexMessage:
 		if err := s.holder.DeleteIndex(obj.Index); err != nil {
 			return err
 		}
-	case *internal.CreateFieldMessage:
+	case *CreateFieldMessage:
 		idx := s.holder.Index(obj.Index)
 		if idx == nil {
 			return fmt.Errorf("Local Index not found: %s", obj.Index)
 		}
-		opt := decodeFieldOptions(obj.Meta)
+		opt := obj.Meta
 		_, err := idx.CreateField(obj.Field, *opt)
 		if err != nil {
 			return err
 		}
-	case *internal.DeleteFieldMessage:
+	case *DeleteFieldMessage:
 		idx := s.holder.Index(obj.Index)
 		if err := idx.DeleteField(obj.Field); err != nil {
 			return err
 		}
-	case *internal.CreateViewMessage:
+	case *CreateViewMessage:
 		f := s.holder.Field(obj.Index, obj.Field)
 		if f == nil {
 			return fmt.Errorf("Local Field not found: %s", obj.Field)
@@ -473,7 +471,7 @@ func (s *Server) receiveMessage(pb proto.Message) error {
 		if err != nil {
 			return err
 		}
-	case *internal.DeleteViewMessage:
+	case *DeleteViewMessage:
 		f := s.holder.Field(obj.Index, obj.Field)
 		if f == nil {
 			return fmt.Errorf("Local Field not found: %s", obj.Field)
@@ -482,36 +480,36 @@ func (s *Server) receiveMessage(pb proto.Message) error {
 		if err != nil {
 			return err
 		}
-	case *internal.ClusterStatus:
-		err := s.cluster.mergeClusterStatus(decodeClusterStatus(obj))
+	case *ClusterStatus:
+		err := s.cluster.mergeClusterStatus(obj)
 		if err != nil {
 			return err
 		}
-	case *internal.ResizeInstruction:
-		err := s.cluster.followResizeInstruction(decodeResizeInstruction(obj))
+	case *ResizeInstruction:
+		err := s.cluster.followResizeInstruction(obj)
 		if err != nil {
 			return err
 		}
-	case *internal.ResizeInstructionComplete:
+	case *ResizeInstructionComplete:
 		err := s.cluster.markResizeInstructionComplete(obj)
 		if err != nil {
 			return err
 		}
-	case *internal.SetCoordinatorMessage:
-		s.cluster.setCoordinator(DecodeNode(obj.New))
-	case *internal.UpdateCoordinatorMessage:
-		s.cluster.updateCoordinator(DecodeNode(obj.New))
-	case *internal.NodeStateMessage:
+	case *SetCoordinatorMessage:
+		s.cluster.setCoordinator(obj.New)
+	case *UpdateCoordinatorMessage:
+		s.cluster.updateCoordinator(obj.New)
+	case *NodeStateMessage:
 		err := s.cluster.receiveNodeState(obj.NodeID, obj.State)
 		if err != nil {
 			return err
 		}
-	case *internal.RecalculateCaches:
+	case *RecalculateCaches:
 		s.holder.RecalculateCaches()
-	case *internal.NodeEventMessage:
-		s.cluster.ReceiveEvent(DecodeNodeEvent(obj))
-	case *internal.NodeStatus:
-		s.handleRemoteStatus(pb)
+	case *nodeEvent:
+		s.cluster.ReceiveEvent(obj)
+	case *NodeStatus:
+		s.handleRemoteStatus(obj)
 	}
 
 	return nil
@@ -519,6 +517,7 @@ func (s *Server) receiveMessage(pb proto.Message) error {
 
 // SendSync represents an implementation of Broadcaster.
 func (s *Server) SendSync(m Message) error {
+	pb := encode(m)
 	var eg errgroup.Group
 	for _, node := range s.cluster.Nodes {
 		node := node
@@ -537,12 +536,13 @@ func (s *Server) SendSync(m Message) error {
 }
 
 // SendAsync represents an implementation of Broadcaster.
-func (s *Server) SendAsync(pb proto.Message) error {
+func (s *Server) SendAsync(m Message) error {
 	return ErrNotImplemented
 }
 
 // SendTo represents an implementation of Broadcaster.
-func (s *Server) SendTo(to *Node, pb proto.Message) error {
+func (s *Server) SendTo(to *Node, m Message) error {
+	pb := encode(m)
 	s.logger.Printf("SendTo: %s", to.URI)
 	return s.defaultClient.SendMessage(context.Background(), &to.URI, pb)
 }
@@ -554,7 +554,7 @@ func (s *Server) node() Node {
 }
 
 // handleRemoteStatus receives incoming NodeStatus from remote nodes.
-func (s *Server) handleRemoteStatus(pb proto.Message) {
+func (s *Server) handleRemoteStatus(pb Message) {
 	// Ignore NodeStatus messages until the cluster is in a Normal state.
 	if s.cluster.State() != ClusterStateNormal {
 		return
@@ -564,16 +564,16 @@ func (s *Server) handleRemoteStatus(pb proto.Message) {
 		// Make sure the holder has opened.
 		<-s.holder.opened
 
-		err := s.mergeRemoteStatus(pb.(*internal.NodeStatus))
+		err := s.mergeRemoteStatus(pb.(*NodeStatus))
 		if err != nil {
 			s.logger.Printf("merge remote status: %s", err)
 		}
 	}()
 }
 
-func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error {
+func (s *Server) mergeRemoteStatus(ns *NodeStatus) error {
 	// Ignore status updates from self.
-	if s.nodeID == DecodeNode(ns.Node).ID {
+	if s.nodeID == ns.Node.ID {
 		return nil
 	}
 
@@ -584,7 +584,7 @@ func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error {
 
 	// Sync maxShards.
 	oldmaxshards := s.holder.maxShards()
-	for index, newMax := range ns.MaxShards.Standard {
+	for index, newMax := range ns.MaxShards {
 		localIndex := s.holder.Index(index)
 		// if we don't know about an index locally, log an error because
 		// indexes should be created and synced prior to shard creation

utils_internal_test.go

@@ -24,7 +24,6 @@ import (
 	"time"
 
 	"github.com/gogo/protobuf/proto"
-	"github.com/pilosa/pilosa/internal"
 )
 
 // NewTestCluster returns a cluster with n nodes and uses a mod-based hasher.
@@ -304,9 +303,9 @@ func (t *ClusterCluster) Close() error {
 }
 
 // SendSync is a test implemenetation of Broadcaster SendSync method.
-func (t *ClusterCluster) SendSync(pb proto.Message) error {
-	switch obj := pb.(type) {
-	case *internal.ClusterStatus:
+func (t *ClusterCluster) SendSync(m Message) error {
+	switch obj := m.(type) {
+	case *ClusterStatus:
 		// Apply the send message to all nodes (except the coordinator).
 		for _, c := range t.Clusters {
 			c.mergeClusterStatus(obj)
@@ -322,19 +321,19 @@ func (t *ClusterCluster) SendSync(pb proto.Message) error {
 }
 
 // SendAsync is a test implemenetation of Broadcaster SendAsync method.
-func (t *ClusterCluster) SendAsync(pb proto.Message) error {
+func (t *ClusterCluster) SendAsync(Message) error {
 	return nil
 }
 
 // SendTo is a test implemenetation of Broadcaster SendTo method.
-func (t *ClusterCluster) SendTo(to *Node, pb proto.Message) error {
-	switch obj := pb.(type) {
-	case *internal.ResizeInstruction:
+func (t *ClusterCluster) SendTo(to *Node, m Message) error {
+	switch obj := m.(type) {
+	case *ResizeInstruction:
 		err := t.FollowResizeInstruction(obj)
 		if err != nil {
 			return err
 		}
-	case *internal.ResizeInstructionComplete:
+	case *ResizeInstructionComplete:
 		coord := t.clusterByID(to.ID)
 		go coord.markResizeInstructionComplete(obj)
 	}
@@ -342,10 +341,10 @@ func (t *ClusterCluster) SendTo(to *Node, pb proto.Message) error {
 }
 
 // FollowResizeInstruction is a version of cluster.FollowResizeInstruction used for testing.
-func (t *ClusterCluster) FollowResizeInstruction(instr *internal.ResizeInstruction) error {
+func (t *ClusterCluster) FollowResizeInstruction(instr *ResizeInstruction) error {
 
 	// Prepare the return message.
-	complete := &internal.ResizeInstructionComplete{
+	complete := &ResizeInstructionComplete{
 		JobID: instr.JobID,
 		Node:  instr.Node,
 		Error: "",
@@ -356,7 +355,7 @@ func (t *ClusterCluster) FollowResizeInstruction(instr *internal.ResizeInstructi
 
 		// figure out which node it was meant for, then call the operation on that cluster
 		// basically need to mimic this: client.RetrieveShardFromURI(context.Background(), src.Index, src.Field, src.View, src.Shard, srcURI)
-		instrNode := DecodeNode(instr.Node)
+		instrNode := instr.Node
 		destCluster := t.clusterByID(instrNode.ID)
 
 		// Sync the schema received in the resize instruction.
@@ -365,7 +364,7 @@ func (t *ClusterCluster) FollowResizeInstruction(instr *internal.ResizeInstructi
 		}
 
 		for _, src := range instr.Sources {
-			srcNode := DecodeNode(src.Node)
+			srcNode := src.Node
 			srcCluster := t.clusterByID(srcNode.ID)
 
 			srcFragment := srcCluster.holder.fragment(src.Index, src.Field, src.View, src.Shard)
@@ -405,6 +404,6 @@ func (t *ClusterCluster) FollowResizeInstruction(instr *internal.ResizeInstructi
 		complete.Error = err.Error()
 	}
 
-	node := DecodeNode(instr.Coordinator)
+	node := instr.Coordinator
 	return t.SendTo(node, complete)
 }