scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-10-09 04:17:51 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

let remote nodes know that its safe to open Holder when launching based on existing taxonomy

Browse source
This commit is contained in:
Travis Turner 2017-11-27 10:32:49 -06:00
parent 013cd0cd95
commit ae63adfaac
No known key found for this signature in database
GPG key ID: 7F08008DFD9314C9
1 changed files with 17 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
cluster.go
View file

@ -301,10 +301,7 @@ func (c *Cluster) SetNodeState(state string) error {
State: state,
}
node := &Node{
URI: c.Coordinator,
}
if err := c.Broadcaster.SendTo(node, ns); err != nil {
if err := c.sendTo(c.Coordinator, ns); err != nil {
return fmt.Errorf("sending node state error: err=%s", err)
}
@ -779,6 +776,14 @@ func (c *Cluster) setStateAndBroadcast(state string) error {
return c.Broadcaster.SendSync(c.Status())
}
func (c *Cluster) sendTo(to URI, msg proto.Message) error {
node := &Node{URI: to}
if err := c.Broadcaster.SendTo(node, msg); err != nil {
return err
}
return nil
}
// ListenForJoins handles cluster-resize events.
func (c *Cluster) ListenForJoins() {
c.wg.Add(1)
@ -876,14 +881,14 @@ func (c *Cluster) generateResizeJobByAction(nodeAction nodeAction) (*ResizeJob,
pbSchema := c.Holder.EncodeSchema()
// multiIndex is a map of sources for each node in toCluster.
// Initialize the map with all the nodes in toCluster.
// multiIndex is a map of sources initialized with all the nodes in toCluster.
multiIndex := make(map[URI][]*internal.ResizeSource)
for _, n := range toCluster.Nodes {
multiIndex[n.URI] = nil
}
// Add to m the instructions for each index.
// Add to multiIndex the instructions for each index.
for _, idx := range c.Holder.Indexes() {
fragSources, err := c.fragSources(toCluster, idx)
if err != nil {
@ -1002,10 +1007,7 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err
complete.Error = err.Error()
}
node := &Node{
URI: decodeURI(instr.Coordinator),
}
if err := c.Broadcaster.SendTo(node, complete); err != nil {
if err := c.sendTo(decodeURI(instr.Coordinator), complete); err != nil {
c.logger().Printf("sending resizeInstructionComplete error: err=%s", err)
}
}()
@ -1401,6 +1403,10 @@ func (c *Cluster) nodeJoin(uri URI) error {
if c.haveTopologyAgreement() && c.allNodesReady() {
return c.setStateAndBroadcast(ClusterStateNormal)
} else {
// Send the status to the remote node. This lets the remote node
// know that it can proceed with opening its Holder.
return c.sendTo(uri, c.Status())
}
return nil