refactor to use multiple feeds

[whyrusleeping]

No description provided.

[keks]

I'm porting Orbit to Go and this would be really helpful. Is there a reason this wasn't merged?

[keks]

I see the code severely diverged. I'll try to make it mergable tomorrow!

[whyrusleeping]

@keks we've decided against doing things this way. Although i'm not against being able to open multiple subsriptions for the same topic, its a little difficult to get working correctly (where the closing of subscriptions get a little sticky)

[keks]

Can we chat about this tomorrow? I don't think it really makes sense this way so I'd like to find a good solution for this.

Edit: I see I misread your post, sorry. I think being able to do that is super important and I would like to know your concerns about this solution so I can figure something out that satisfies all conditions. Will you be on IRC tomorrow (Tuesday) at 17:00 UTC? If not, please propose some later time.

Thanks!

[whyrusleeping]

@keks i'll likely be online at that time. Thats 9am for me, so its a tad early, but i've been waking up earlier lately and it should be fine.

[keks]

Here are my thoughts. What do you think about this?

Note: The version I currently use is the feat/feed-refactor branch, rebased on top of master. you can find it here.

Currently Subscribe takes a context.Context and uses that for cancelling the subscription. That means the caller first needs to prepare a cancellable context and pass that to Subscribe. Usually cancellable contexts are used to actually cancel the called, blocking function. Subscribe doesn't block though.

I propose that Subscribe doesn't return a <-chan *Message but a Subscription like this:

type Subscription struct {
  toptic string
  ch <-chan *Message
  cancelCh chan *Subscription
  err error
}

func (sub *Subscription) Topic() string {
  return sub.topic
}

func (sub *Subscription) Next() (*Message, error) {
  msg, ok := <- sub.ch

  if !ok {
    return msg, err
  }

  return msg, nil
}

func (sub *Subscription) Cancel() {
  sub.cancelCh <- sub
}

That way you could use

sub, err := p.Subscribe(topic)
if err != nil { ... }
defer sub.Cancel()

as an idiom.

p.Subscribe works like it did before, except it returns a *Subscription. However, p.GetFeed would look a bit different:

func (p *PubSub) GetFeed(topic string) (*Subscription, error) {
    out := make(chan *Subscription, 1)
    p.addFeedHook <- &addFeedReq{
        topic: topic,
        resp:  out,
    }

    resp := <-out
    if resp == nil {
        return nil, fmt.Errorf("not subscribed to topic %s", topic)
    }
    return resp, nil
}

and the process loop would be changed like this:

      case sub := <- p.cancelCh:
        subs, ok := p.myTopics[req.topic]
        if !ok {
          continue
        }

        delete(subs, sub)
        if len(subs) == 0 { ... } // remove lower level subscription
      case req := <-p.addFeedHook:
        subs, ok := p.myTopics[req.topic]
        sub := &Subscription{
          ch: make(chan *Message),
          topic: req.topic,
          cancelCh: p.cancelCh,
        }
        if ok {
          subs[sub] = stuct{}{}
        }

        req.resp <- sub
      case ... //and so on

Note that I added cancelCh to PubSub and changed p.myTopics from map[string][]chan Message to map[string]map[*Subscription]struct{}. map[foo]struct{} is a common way to implement sets and you don't need to do the cleanup stuff you do here; instead you can simple delete the *Subscription.
This obviously requires changes all over the file, but well, that's life.

Wow, I guess I just could have changed all of this directly in the code, but I hope I get my intentions across better this way. If you sign off on this copypasta will be my friend :)

[whyrusleeping]

@keks that looks pretty good to me. Good job on writing the whole PR in this PR :P

[keks]

@whyrusleeping this can be closed as #10 is merged.