refactor(event): wiring watchfs up to FSI events

api/websocket.go

@@ -31,7 +31,7 @@ const (
 func (s Server) ServeWebsocket(ctx context.Context) {
  // Watch the filesystem. Events will be sent to websocket connections.
  node := s.Node()
- fsmessages, err := s.startFilesysWatcher(node)
+ fsmessages, err := s.startFilesysWatcher(ctx, node)
  if err != nil {
  log.Infof("Watching filesystem error: %s", err)
  return
@@ -113,7 +113,7 @@ func (s Server) ServeWebsocket(ctx context.Context) {
  }()
 }
 
-func (s Server) startFilesysWatcher(node *p2p.QriNode) (chan watchfs.FilesysEvent, error) {
+func (s Server) startFilesysWatcher(ctx context.Context, node *p2p.QriNode) (chan watchfs.FilesysEvent, error) {
  refs, err := node.Repo.References(0, 100)
  if err != nil {
  return nil, err
@@ -131,7 +131,7 @@ func (s Server) startFilesysWatcher(node *p2p.QriNode) (chan watchfs.FilesysEven
  }
  // Watch those paths.
  // TODO(dlong): When datasets are removed or renamed update the watchlist.
- s.Instance.Watcher = watchfs.NewFilesysWatcher()
+ s.Instance.Watcher = watchfs.NewFilesysWatcher(ctx, s.Instance.Bus())
  fsmessages := s.Instance.Watcher.Begin(paths)
  return fsmessages, nil
 }

dscache/dscache.go

@@ -178,12 +178,13 @@ func (d *Dscache) ListRefs() ([]reporef.DatasetRef, error) {
 }
 
 func (d *Dscache) update(act *logbook.Action) {
- if act.Type == logbook.ActionDatasetNameInit {
- if err := d.updateInitDataset(act); err != nil {
+ switch act.Type {
+ case logbook.ActionDatasetNameInit:
+ if err := d.updateInitDataset(act); err != nil && err != ErrNoDscache {
  log.Error(err)
  }
- } else if act.Type == logbook.ActionDatasetChange {
- if err := d.updateMoveCursor(act); err != nil {
+ case logbook.ActionDatasetChange:
+ if err := d.updateMoveCursor(act); err != nil && err != ErrNoDscache {
  log.Error(err)
  }
  }

event/event.go

@@ -23,6 +23,19 @@ type Event struct {
  Payload interface{}
 }
 
+// Publisher is an interface that can only publish an event
+type Publisher interface {
+ Publish(t Topic, data interface{})
+}
+
+// NilPublisher replaces a nil value, does nothing
+type NilPublisher struct {
+}
+
+// Publish does nothing with the event
+func (n *NilPublisher) Publish(t Topic, data interface{}) {
+}
+
 // Bus is a central coordination point for event publication and subscription
 // zero or more subscribers register topics to be notified of, a publisher
 // writes a topic event to the bus, which broadcasts to all subscribers of that
@@ -32,11 +45,15 @@ type Bus interface {
  Publish(t Topic, data interface{})
  // Subscribe to one or more topics
  Subscribe(topics ...Topic) <-chan Event
+ // Unsubscribe cleans up a channel that no longer need to receive events
+ Unsubscribe(<-chan Event)
  // SubscribeOnce to one or more topics. the returned channel will only fire
  // once, when the first event that matches any of the given topics
  // the common use case for multiple subscriptions is subscribing to both
  // success and error events
  SubscribeOnce(types ...Topic) <-chan Event
+ // NumSubscriptions returns the number of subscribers to the bus's events
+ NumSubscribers() int
 }
 
 type dataChannels []chan Event
@@ -134,6 +151,30 @@ func (b *bus) Subscribe(topics ...Topic) <-chan Event {
  return ch
 }
 
+// Unsubscribe cleans up a channel that no longer need to receive events
+func (b *bus) Unsubscribe(unsub <-chan Event) {
+ for topic, channels := range b.subs {
+ var replace dataChannels
+ for i, ch := range channels {
+ if ch == unsub {
+ replace = append(channels[:i], channels[i+1:]...)
+ }
+ }
+ if replace != nil {
+ b.subs[topic] = replace
+ }
+ }
+}
+
+// NumSubscribers returns the number of subscribers to the bus's events
+func (b *bus) NumSubscribers() int {
+ total := 0
+ for _, channels := range b.subs {
+ total += len(channels)
+ }
+ return total
+}
+
 // SubscribeOnce will only get one event of the topic, then close itself
 func (b *bus) SubscribeOnce(topics ...Topic) <-chan Event {
  b.onceLk.Lock()

event/event_test.go

@@ -3,6 +3,7 @@ package event
 import (
  "context"
  "fmt"
+ "testing"
 )
 
 func Example() {
@@ -51,3 +52,28 @@ func Example() {
  // hello
  // it didn't work?
 }
+
+func TestSubscribeUnsubscribe(t *testing.T) {
+ ctx := context.Background()
+ const testTopic = Topic("test_event")
+
+ b := NewBus(ctx)
+ ch1 := b.Subscribe(testTopic)
+ ch2 := b.Subscribe(testTopic)
+
+ if b.NumSubscribers() != 2 {
+ t.Errorf("expected 2 subscribers, got %d", b.NumSubscribers())
+ }
+
+ b.Unsubscribe(ch1)
+
+ if b.NumSubscribers() != 1 {
+ t.Errorf("expected 1 subscribers, got %d", b.NumSubscribers())
+ }
+
+ b.Unsubscribe(ch2)
+
+ if b.NumSubscribers() != 0 {
+ t.Errorf("expected 1 subscribers, got %d", b.NumSubscribers())
+ }
+}

fsi/fsi.go

@@ -21,6 +21,7 @@ import (
  golog "github.com/ipfs/go-log"
  "github.com/qri-io/qri/base"
  "github.com/qri-io/qri/base/component"
+ "github.com/qri-io/qri/event"
  "github.com/qri-io/qri/repo"
  reporef "github.com/qri-io/qri/repo/ref"
 )
@@ -60,11 +61,15 @@ func RepoPath(repoPath string) string {
 type FSI struct {
  // repository for resolving dataset names
  repo repo.Repo
+ pub event.Publisher
 }
 
 // NewFSI creates an FSI instance from a path to a links flatbuffer file
-func NewFSI(r repo.Repo) *FSI {
- return &FSI{repo: r}
+func NewFSI(r repo.Repo, pub event.Publisher) *FSI {
+ if pub == nil {
+ pub = &event.NilPublisher{}
+ }
+ return &FSI{repo: r, pub: pub}
 }
 
 // LinkedRefs returns a list of linked datasets and their connected directories
@@ -143,6 +148,13 @@ func (fsi *FSI) CreateLink(dirPath, refStr string) (alias string, rollback func(
  removeRefFunc()
  }
 
+ // Send an event to the bus about this checkout
+ fsi.pub.Publish(event.ETFSICreateLinkEvent, event.FSICreateLinkEvent{
+ FSIPath: dirPath,
+ Username: ref.Peername,
+ Dsname: ref.Name,
+ })
+
  return ref.AliasString(), removeLinkAndRemoveRefFunc, err
 }
 

fsi/fsi_test.go

@@ -55,7 +55,7 @@ func TestCreateLink(t *testing.T) {
  paths := NewTmpPaths()
  defer paths.Close()
 
- fsi := NewFSI(paths.testRepo)
+ fsi := NewFSI(paths.testRepo, nil)
  link, _, err := fsi.CreateLink(paths.firstDir, "me/test_ds")
  if err != nil {
  t.Fatalf(err.Error())
@@ -89,7 +89,7 @@ func TestCreateLinkTwice(t *testing.T) {
  paths := NewTmpPaths()
  defer paths.Close()
 
- fsi := NewFSI(paths.testRepo)
+ fsi := NewFSI(paths.testRepo, nil)
  _, _, err := fsi.CreateLink(paths.firstDir, "me/test_ds")
  if err != nil {
  t.Fatalf(err.Error())
@@ -142,7 +142,7 @@ func TestCreateLinkAlreadyLinked(t *testing.T) {
  paths := NewTmpPaths()
  defer paths.Close()
 
- fsi := NewFSI(paths.testRepo)
+ fsi := NewFSI(paths.testRepo, nil)
  _, _, err := fsi.CreateLink(paths.firstDir, "me/test_ds")
  if err != nil {
  t.Fatalf(err.Error())
@@ -163,7 +163,7 @@ func TestCreateLinkAgainOnceQriRefRemoved(t *testing.T) {
  paths := NewTmpPaths()
  defer paths.Close()
 
- fsi := NewFSI(paths.testRepo)
+ fsi := NewFSI(paths.testRepo, nil)
  _, _, err := fsi.CreateLink(paths.firstDir, "me/test_ds")
  if err != nil {
  t.Fatalf(err.Error())
@@ -205,7 +205,7 @@ func TestModifyLinkReference(t *testing.T) {
  paths := NewTmpPaths()
  defer paths.Close()
 
- fsi := NewFSI(paths.testRepo)
+ fsi := NewFSI(paths.testRepo, nil)
  _, _, err := fsi.CreateLink(paths.firstDir, "me/test_ds")
  if err != nil {
  t.Fatal(err)
@@ -248,7 +248,7 @@ func TestModifyLinkDirectory(t *testing.T) {
  paths := NewTmpPaths()
  defer paths.Close()
 
- fsi := NewFSI(paths.testRepo)
+ fsi := NewFSI(paths.testRepo, nil)
  _, _, err := fsi.CreateLink(paths.firstDir, "me/another_dataset")
  if err != nil {
  t.Fatal(err)
@@ -277,7 +277,7 @@ func TestUnlink(t *testing.T) {
  paths := NewTmpPaths()
  defer paths.Close()
 
- fsi := NewFSI(paths.testRepo)
+ fsi := NewFSI(paths.testRepo, nil)
  _, _, err := fsi.CreateLink(paths.firstDir, "me/test_ds")
  if err != nil {
  t.Fatalf(err.Error())

fsi/status_test.go

@@ -37,7 +37,7 @@ func TestStatusValid(t *testing.T) {
  paths := NewTmpPaths()
  defer paths.Close()
 
- fsi := NewFSI(paths.testRepo)
+ fsi := NewFSI(paths.testRepo, nil)
  _, _, err := fsi.CreateLink(paths.firstDir, "me/test_ds")
  if err != nil {
  t.Fatalf(err.Error())
@@ -81,7 +81,7 @@ func TestStatusInvalidDataset(t *testing.T) {
  paths := NewTmpPaths()
  defer paths.Close()
 
- fsi := NewFSI(paths.testRepo)
+ fsi := NewFSI(paths.testRepo, nil)
  _, _, err := fsi.CreateLink(paths.firstDir, "me/test_ds")
  if err != nil {
  t.Fatalf(err.Error())
@@ -105,7 +105,7 @@ func TestStatusInvalidMeta(t *testing.T) {
  paths := NewTmpPaths()
  defer paths.Close()
 
- fsi := NewFSI(paths.testRepo)
+ fsi := NewFSI(paths.testRepo, nil)
  _, _, err := fsi.CreateLink(paths.firstDir, "me/test_ds")
  if err != nil {
  t.Fatalf(err.Error())
@@ -127,7 +127,7 @@ func TestStatusNotFound(t *testing.T) {
  paths := NewTmpPaths()
  defer paths.Close()
 
- fsi := NewFSI(paths.testRepo)
+ fsi := NewFSI(paths.testRepo, nil)
  _, _, err := fsi.CreateLink(paths.firstDir, "me/test_ds")
  if err != nil {
  t.Fatalf(err.Error())

lib/fsi.go

@@ -10,7 +10,6 @@ import (
  "github.com/qri-io/qri/base"
  "github.com/qri-io/qri/base/component"
  "github.com/qri-io/qri/base/dsfs"
- "github.com/qri-io/qri/event"
  "github.com/qri-io/qri/fsi"
  "github.com/qri-io/qri/repo"
  reporef "github.com/qri-io/qri/repo/ref"
@@ -190,14 +189,6 @@ func (m *FSIMethods) Checkout(p *CheckoutParams, out *string) (err error) {
  }
  log.Debugf("Checkout wrote components, successfully checked out dataset")
 
- // Send an event to the bus about this checkout
- m.inst.Bus().Publish(event.ETFSICreateLinkEvent, event.FSICreateLinkEvent{
- FSIPath: p.Dir,
- Username: ref.Peername,
- Dsname: ref.Name,
- })
- log.Debugf("Checkout published an event")
-
  log.Debugf("Checkout successfully checked out dataset")
  return nil
 }

lib/lib.go

@@ -302,6 +302,7 @@ func NewInstance(ctx context.Context, repoPath string, opts ...Option) (qri *Ins
  streams: o.Streams,
  registry: o.regclient,
  logbook: o.logbook,
+ bus: event.NewBus(ctx),
  }
  qri = inst
 
@@ -371,10 +372,6 @@ func NewInstance(ctx context.Context, repoPath string, opts ...Option) (qri *Ins
  }
  }
 
- if inst.bus == nil {
- inst.bus = newEventBus(ctx)
- }
-
  if inst.registry == nil {
  inst.registry = newRegClient(ctx, cfg)
  }
@@ -400,7 +397,7 @@ func NewInstance(ctx context.Context, repoPath string, opts ...Option) (qri *Ins
  // were somewhere else we could move it there
  _ = base.SetFileHidden(inst.repoPath)
 
- inst.fsi = fsi.NewFSI(inst.repo)
+ inst.fsi = fsi.NewFSI(inst.repo, inst.bus)
  }
 
  if inst.node == nil {
@@ -590,8 +587,8 @@ func NewInstanceFromConfigAndNode(cfg *config.Config, node *p2p.QriNode) *Instan
  inst.repo = node.Repo
  inst.store = node.Repo.Store()
  inst.qfs = node.Repo.Filesystem()
- inst.fsi = fsi.NewFSI(inst.repo)
  inst.bus = event.NewBus(ctx)
+ inst.fsi = fsi.NewFSI(inst.repo, inst.bus)
  }
 
  return inst