block sent listener is not called for every block sent

[ipfs]

block sent listener is not called for every block sent

[dirkmc]

It seems that the root cause of the issue is that when the messagequeue sends a message, the message may contain several blocks, but only one "Sent" event is emitted.

go-graphsync/messagequeue/messagequeue.go

Lines 202 to 207 in 11d30c6

	func (mq *MessageQueue) attemptSendAndRecovery(message gsmsg.GraphSyncMessage, topic Topic) bool {
	err := mq.sender.SendMsg(mq.ctx, message)
	if err == nil {
	mq.eventPublisher.Publish(topic, Event{Name: Sent})
	return true
	}

When this event bubbles up to the responsemanager subscriber, it only includes the block data for the first block (not all blocks in the message)
https://github.com/ipfs/go-graphsync/blob/master/responsemanager/subscriber.go#L42

[hannahhoward]

Hey, so this exposes a flaw in the mappable subscriber system, and provides a pretext for refactoring the interface.

[hannahhoward]

@acruikshank I am recommending a fix to the way MappableSubscriber currently works, as well as some interface changes that will make things more clear:

These are my recommended new interfaces -- please don't take these set in stone and please feel free to improve naming. (taken from: https://github.com/ipfs/go-graphsync/blob/master/notifications/types.go)

---

// Topic is a topic that events appear on
type Topic interface{}

// Event is a publishable event
type Event interface{}

// TopicData is data added to every message broadcast on a topic
type TopicData interface{}

// Subscriber is a subscriber that can receive events
type Subscriber interface {
	OnNext(Topic, Event, ...TopicData)
	OnClose(Topic, ...TopicData)
}

// TopicDataSubscriber is a subscriber that can inject data to into subscriber callbacks when events happen for a given topic
type TopicDataSubscriber interface {
	Subscriber
	AddTopicData(topic Topic, data TopicData)
}

// Subscribable is a stream that can be subscribed to
type Subscribable interface {
	Subscribe(topic Topic, sub Subscriber) bool
	Unsubscribe(sub Subscriber) bool
}

// Publisher is an publisher of events that can be subscribed to
type Publisher interface {
	Close(Topic)
	Publish(Topic, Event)
	Shutdown()
	Startup()
	Subscribable
}

// EventTransform if a fucntion transforms one kind of event to another
type EventTransform func(Event) Event

// Notifee is a topic data subscriber plus a set of data you want to add to any topics subscribed to
// (used to call SubscribeWithData to inject data when events for a given topic emit)
type Notifee struct {
	Data      TopicData
	Subscriber TopicDataSubscriber
}

// SubscribeWithData subscribes to the given subscriber on the given topic, and adds the notifiees
// custom data into the list of data injected into callbacks when events occur on that topic
func SubscribeWithData(p Subscribable, topic Topic, notifee Notifee) {
	notifee.Subscriber.AddTopicData(topic, notifee.Data)
	p.Subscribe(topic, notifee.Subscriber)
}

--

I also recommend:

• NewMappableSubscribe -> NewTopicDataSubscriber
• Lose the event transform parameter to NewTopicDataSubscriber -- we never use this other than to pass a function that does no transformation
• change the internal map from map[Topic]Topic to map[Topic][]TopicData -- note the fact that it's currently over written is actually the root cause of the issue

--

All of this will neccesitate changes to all of the subscribers. We will need to dispatch multiple BlockSent notifications in the subscriber for the responsemanager, which will ultimately be the solution to the root issue.