Accept/Reject requests up front

graphsync.go

@@ -277,6 +277,7 @@ type BlockData interface {
 // IncomingRequestHookActions are actions that a request hook can take to change
 // behavior for the response
 type IncomingRequestHookActions interface {
+ AugmentContext(func(reqCtx context.Context) context.Context)
  SendExtensionData(ExtensionData)
  UsePersistenceOption(name string)
  UseLinkTargetNodePrototypeChooser(traversal.LinkTargetNodePrototypeChooser)
@@ -323,16 +324,6 @@ type RequestUpdatedHookActions interface {
  UnpauseResponse()
 }
 
-// RequestQueuedHookActions are actions that can be taken in a request queued hook to
-// change execution of the response
-type RequestQueuedHookActions interface {
- AugmentContext(func(reqCtx context.Context) context.Context)
-}
-
-// OnIncomingRequestQueuedHook is a hook that runs each time a new incoming request is added to the responder's task queue.
-// It receives the peer that sent the request and all data about the request.
-type OnIncomingRequestQueuedHook func(p peer.ID, request RequestData, hookActions RequestQueuedHookActions)
-
 // OnIncomingRequestHook is a hook that runs each time a new request is received.
 // It receives the peer that sent the request and all data about the request.
 // It receives an interface for customizing the response to this request
@@ -381,9 +372,9 @@ type OnReceiverNetworkErrorListener func(p peer.ID, err error)
 // OnResponseCompletedListener provides a way to listen for when responder has finished serving a response
 type OnResponseCompletedListener func(p peer.ID, request RequestData, status ResponseStatusCode)
 
-// OnOutgoingRequestProcessingListener is called when a request actually begins processing (reaches
-// the top of the outgoing request queue)
-type OnOutgoingRequestProcessingListener func(p peer.ID, request RequestData, inProgressRequestCount int)
+// OnRequestProcessingListener is called when a request actually begins processing (reaches
+// the top of the request queue)
+type OnRequestProcessingListener func(p peer.ID, request RequestData, inProgressRequestCount int)
 
 // OnRequestorCancelledListener provides a way to listen for responses the requestor canncels
 type OnRequestorCancelledListener func(p peer.ID, request RequestData)
@@ -480,9 +471,6 @@ type GraphExchange interface {
  // UnregisterPersistenceOption unregisters an alternate loader/storer combo
  UnregisterPersistenceOption(name string) error
 
- // RegisterIncomingRequestQueuedHook adds a hook that runs when a new incoming request is added to the responder's task queue.
- RegisterIncomingRequestQueuedHook(hook OnIncomingRequestQueuedHook) UnregisterHookFunc
-
  // RegisterIncomingRequestHook adds a hook that runs when a request is received
  RegisterIncomingRequestHook(hook OnIncomingRequestHook) UnregisterHookFunc
 
@@ -501,9 +489,13 @@ type GraphExchange interface {
  // RegisterRequestUpdatedHook adds a hook that runs every time an update to a request is received
  RegisterRequestUpdatedHook(hook OnRequestUpdatedHook) UnregisterHookFunc
 
- // RegisterOutgoingRequestProcessingListener adds a listener that gets called when a request actually begins processing (reaches
+ // RegisterOutgoingRequestProcessingListener adds a listener that gets called when an outgoing request actually begins processing (reaches
+ // the top of the outgoing request queue)
+ RegisterOutgoingRequestProcessingListener(listener OnRequestProcessingListener) UnregisterHookFunc
+
+ // RegisterOutgoingRequestProcessingListener adds a listener that gets called when an incoming request actually begins processing (reaches
  // the top of the outgoing request queue)
- RegisterOutgoingRequestProcessingListener(listener OnOutgoingRequestProcessingListener) UnregisterHookFunc
+ RegisterIncomingRequestProcessingListener(listener OnRequestProcessingListener) UnregisterHookFunc
 
  // RegisterCompletedResponseListener adds a listener on the responder for completed responses
  RegisterCompletedResponseListener(listener OnResponseCompletedListener) UnregisterHookFunc

impl/graphsync.go

@@ -56,11 +56,11 @@ type GraphSync struct {
  requestExecutor *executor.Executor
  responseAssembler *responseassembler.ResponseAssembler
  peerManager *peermanager.PeerMessageManager
- incomingRequestQueuedHooks *responderhooks.IncomingRequestQueuedHooks
  incomingRequestHooks *responderhooks.IncomingRequestHooks
  outgoingBlockHooks *responderhooks.OutgoingBlockHooks
  requestUpdatedHooks *responderhooks.RequestUpdatedHooks
- outgoingRequestProcessingListeners *listeners.OutgoingRequestProcessingListeners
+ incomingRequestProcessingListeners *listeners.RequestProcessingListeners
+ outgoingRequestProcessingListeners *listeners.RequestProcessingListeners
  completedResponseListeners *listeners.CompletedResponseListeners
  requestorCancelledListeners *listeners.RequestorCancelledListeners
  blockSentListeners *listeners.BlockSentListeners
@@ -224,9 +224,9 @@ func New(parent context.Context, network gsnet.GraphSyncNetwork,
  incomingBlockHooks := requestorhooks.NewBlockHooks()
  networkErrorListeners := listeners.NewNetworkErrorListeners()
  receiverErrorListeners := listeners.NewReceiverNetworkErrorListeners()
- outgoingRequestProcessingListeners := listeners.NewOutgoingRequestProcessingListeners()
+ outgoingRequestProcessingListeners := listeners.NewRequestProcessingListeners()
+ incomingRequestProcessingListeners := listeners.NewRequestProcessingListeners()
  persistenceOptions := persistenceoptions.New()
- requestQueuedHooks := responderhooks.NewRequestQueuedHooks()
  incomingRequestHooks := responderhooks.NewRequestHooks(persistenceOptions)
  outgoingBlockHooks := responderhooks.NewBlockHooks()
  requestUpdatedHooks := responderhooks.NewUpdateHooks()
@@ -255,7 +255,7 @@ func New(parent context.Context, network gsnet.GraphSyncNetwork,
  ctx,
  linkSystem,
  responseAssembler,
- requestQueuedHooks,
+ incomingRequestProcessingListeners,
  incomingRequestHooks,
  requestUpdatedHooks,
  completedResponseListeners,
@@ -273,32 +273,33 @@ func New(parent context.Context, network gsnet.GraphSyncNetwork,
  requestUpdatedHooks,
  )
  graphSync := &GraphSync{
- network: network,
- linkSystem: linkSystem,
- requestManager: requestManager,
- responseManager: responseManager,
- queryExecutor: queryExecutor,
- responseQueue: responseQueue,
- requestQueue: requestQueue,
- requestExecutor: requestExecutor,
- responseAssembler: responseAssembler,
- peerManager: peerManager,
- incomingRequestQueuedHooks: requestQueuedHooks,
- incomingRequestHooks: incomingRequestHooks,
- outgoingBlockHooks: outgoingBlockHooks,
- requestUpdatedHooks: requestUpdatedHooks,
- completedResponseListeners: completedResponseListeners,
- requestorCancelledListeners: requestorCancelledListeners,
- blockSentListeners: blockSentListeners,
- networkErrorListeners: networkErrorListeners,
- receiverErrorListeners: receiverErrorListeners,
- incomingResponseHooks: incomingResponseHooks,
- outgoingRequestHooks: outgoingRequestHooks,
- incomingBlockHooks: incomingBlockHooks,
- persistenceOptions: persistenceOptions,
- ctx: ctx,
- cancel: cancel,
- responseAllocator: responseAllocator,
+ network: network,
+ linkSystem: linkSystem,
+ requestManager: requestManager,
+ responseManager: responseManager,
+ queryExecutor: queryExecutor,
+ responseQueue: responseQueue,
+ requestQueue: requestQueue,
+ requestExecutor: requestExecutor,
+ responseAssembler: responseAssembler,
+ peerManager: peerManager,
+ incomingRequestProcessingListeners: incomingRequestProcessingListeners,
+ outgoingRequestProcessingListeners: outgoingRequestProcessingListeners,
+ incomingRequestHooks: incomingRequestHooks,
+ outgoingBlockHooks: outgoingBlockHooks,
+ requestUpdatedHooks: requestUpdatedHooks,
+ completedResponseListeners: completedResponseListeners,
+ requestorCancelledListeners: requestorCancelledListeners,
+ blockSentListeners: blockSentListeners,
+ networkErrorListeners: networkErrorListeners,
+ receiverErrorListeners: receiverErrorListeners,
+ incomingResponseHooks: incomingResponseHooks,
+ outgoingRequestHooks: outgoingRequestHooks,
+ incomingBlockHooks: incomingBlockHooks,
+ persistenceOptions: persistenceOptions,
+ ctx: ctx,
+ cancel: cancel,
+ responseAllocator: responseAllocator,
  }
 
  requestManager.SetDelegate(peerManager)
@@ -335,8 +336,8 @@ func (gs *GraphSync) RegisterIncomingRequestHook(hook graphsync.OnIncomingReques
 
 // RegisterIncomingRequestQueuedHook adds a hook that runs when a new incoming request is added
 // to the responder's task queue.
-func (gs *GraphSync) RegisterIncomingRequestQueuedHook(hook graphsync.OnIncomingRequestQueuedHook) graphsync.UnregisterHookFunc {
- return gs.incomingRequestQueuedHooks.Register(hook)
+func (gs *GraphSync) RegisterIncomingRequestProcessingListener(listener graphsync.OnRequestProcessingListener) graphsync.UnregisterHookFunc {
+ return gs.incomingRequestProcessingListeners.Register(listener)
 }
 
 // RegisterIncomingResponseHook adds a hook that runs when a response is received
@@ -371,7 +372,7 @@ func (gs *GraphSync) RegisterRequestUpdatedHook(hook graphsync.OnRequestUpdatedH
 
 // RegisterOutgoingRequestProcessingListener adds a listener that gets called when a request actually begins processing (reaches
 // the top of the outgoing request queue)
-func (gs *GraphSync) RegisterOutgoingRequestProcessingListener(listener graphsync.OnOutgoingRequestProcessingListener) graphsync.UnregisterHookFunc {
+func (gs *GraphSync) RegisterOutgoingRequestProcessingListener(listener graphsync.OnRequestProcessingListener) graphsync.UnregisterHookFunc {
  return gs.outgoingRequestProcessingListeners.Register(listener)
 }