Refactor events handlers

benchmarks/testinstance/testinstance.go

@@ -164,7 +164,7 @@ func NewInstance(ctx context.Context, net tn.Network, tempDir string, diskBasedD
 
 	linkSystem := storeutil.LinkSystemForBlockstore(bstore)
 	gs := gsimpl.New(ctx, gsNet, linkSystem, gsimpl.RejectAllRequestsByDefault())
-	transport := gstransport.NewTransport(p, gs, dtNet)
+	transport := gstransport.NewTransport(gs, dtNet)
 	dt, err := dtimpl.NewDataTransfer(namespace.Wrap(dstore, datastore.NewKey("/data-transfers/transfers")), p, transport)
 	if err != nil {
 		return Instance{}, err

channels/channel_state.go

@@ -50,22 +50,22 @@ func (c channelState) Voucher() datatransfer.TypedVoucher {
 	return datatransfer.TypedVoucher{Voucher: ev.Voucher.Node, Type: ev.Type}
 }
 
-// ReceivedCidsTotal returns the number of (non-unique) cids received so far
-// on the channel - note that a block can exist in more than one place in the DAG
-func (c channelState) ReceivedCidsTotal() int64 {
-	return c.ic.ReceivedBlocksTotal
+// ReceivedIndex returns the index, a transport specific identifier for "where"
+// we are in receiving data for a transfer
+func (c channelState) ReceivedIndex() datamodel.Node {
+	return c.ic.ReceivedIndex.Node
 }
 
-// QueuedCidsTotal returns the number of (non-unique) cids queued so far
-// on the channel - note that a block can exist in more than one place in the DAG
-func (c channelState) QueuedCidsTotal() int64 {
-	return c.ic.QueuedBlocksTotal
+// QueuedIndex returns the index, a transport specific identifier for "where"
+// we are in queing data for a transfer
+func (c channelState) QueuedIndex() datamodel.Node {
+	return c.ic.QueuedIndex.Node
 }
 
-// SentCidsTotal returns the number of (non-unique) cids sent so far
-// on the channel - note that a block can exist in more than one place in the DAG
-func (c channelState) SentCidsTotal() int64 {
-	return c.ic.SentBlocksTotal
+// SentIndex returns the index, a transport specific identifier for "where"
+// we are in sending data for a transfer
+func (c channelState) SentIndex() datamodel.Node {
+	return c.ic.SentIndex.Node
 }
 
 // Sender returns the peer id for the node that is sending data
@@ -139,6 +139,25 @@ func (c channelState) RequiresFinalization() bool {
 	return c.ic.RequiresFinalization
 }
 
+func (c channelState) InitiatorPaused() bool {
+	return c.ic.InitiatorPaused
+}
+
+func (c channelState) ResponderPaused() bool {
+	return c.ic.ResponderPaused || c.ic.Status == datatransfer.Finalizing
+}
+
+func (c channelState) BothPaused() bool {
+	return c.InitiatorPaused() && c.ResponderPaused()
+}
+
+func (c channelState) SelfPaused() bool {
+	if c.ic.SelfPeer == c.ic.Initiator {
+		return c.InitiatorPaused()
+	}
+	return c.ResponderPaused()
+}
+
 // Stages returns the current ChannelStages object, or an empty object.
 // It is unsafe for the caller to modify the return value, and changes may not
 // be persisted. It should be treated as immutable.