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

moved the broadcast handler receive message process to the handler ProcessClusterMessage

Browse source
This commit is contained in:
Michael Baird 2017-11-29 15:12:05 -06:00 • committed by Travis Turner
parent 814bc40a16
commit 04790f70f2
No known key found for this signature in database
GPG key ID: 7F08008DFD9314C9
1 changed files with 18 additions and 65 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

83
server.go
View file

@ -36,6 +36,7 @@ import (
"github.com/pilosa/pilosa/diagnostics"
"github.com/pilosa/pilosa/internal"
"golang.org/x/net/context"
"golang.org/x/sync/errgroup"
)
// Default server settings.
@ -332,76 +333,28 @@ func (s *Server) monitorMaxSlices() {
// ReceiveMessage represents an implementation of BroadcastHandler.
func (s *Server) ReceiveMessage(pb proto.Message) error {
switch obj := pb.(type) {
case *internal.CreateSliceMessage:
idx := s.Holder.Index(obj.Index)
if idx == nil {
return fmt.Errorf("Local Index not found: %s", obj.Index)
}
if obj.IsInverse {
idx.SetRemoteMaxInverseSlice(obj.Slice)
} else {
idx.SetRemoteMaxSlice(obj.Slice)
}
case *internal.CreateIndexMessage:
opt := IndexOptions{
ColumnLabel: obj.Meta.ColumnLabel,
TimeQuantum: TimeQuantum(obj.Meta.TimeQuantum),
}
_, err := s.Holder.CreateIndex(obj.Index, opt)
return s.Handler.ProcessClusterMessage(pb)
}
// SendSync represents an implementation of BroadcastHandler.
func (s *Server) SendSync(pb proto.Message) error {
var eg errgroup.Group
for _, node := range s.Cluster.Nodes {
uri, err := node.URI()
if err != nil {
return err
}
case *internal.DeleteIndexMessage:
if err := s.Holder.DeleteIndex(obj.Index); err != nil {
return err
}
case *internal.CreateFrameMessage:
idx := s.Holder.Index(obj.Index)
if idx == nil {
return fmt.Errorf("Local Index not found: %s", obj.Index)
}
opt := FrameOptions{
RowLabel: obj.Meta.RowLabel,
InverseEnabled: obj.Meta.InverseEnabled,
RangeEnabled: obj.Meta.RangeEnabled,
CacheType: obj.Meta.CacheType,
CacheSize: obj.Meta.CacheSize,
TimeQuantum: TimeQuantum(obj.Meta.TimeQuantum),
Fields: decodeFields(obj.Meta.Fields),
}
_, err := idx.CreateFrame(obj.Frame, opt)
if err != nil {
return err
}
case *internal.DeleteFrameMessage:
idx := s.Holder.Index(obj.Index)
if err := idx.DeleteFrame(obj.Frame); err != nil {
return err
}
case *internal.CreateInputDefinitionMessage:
idx := s.Holder.Index(obj.Index)
if idx == nil {
return fmt.Errorf("Local Index not found: %s", obj.Index)
}
idx.CreateInputDefinition(obj.Definition)
case *internal.DeleteInputDefinitionMessage:
idx := s.Holder.Index(obj.Index)
err := idx.DeleteInputDefinition(obj.Name)
if err != nil {
return err
}
case *internal.DeleteViewMessage:
f := s.Holder.Frame(obj.Index, obj.Frame)
if f == nil {
return fmt.Errorf("Local Frame not found: %s", obj.Frame)
}
err := f.DeleteView(obj.View)
if err != nil {
return err
// Don't forward the message to ourselves
if *s.URI != *uri {
ctx := context.WithValue(context.Background(), "uri", uri)
eg.Go(func() error {
return s.defaultClient.ClusterMessage(ctx, pb)
})
}
}
return nil
return eg.Wait()
}
// LocalStatus returns the state of the local node as well as the