datatransfer

package module
v2.0.0-rc8 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 16, 2024 License: Apache-2.0, MIT Imports: 14 Imported by: 89

README

go-data-transfer

CircleCI codecov

A go module to perform data transfers over ipfs/go-graphsync

Description

This module encapsulates protocols for exchanging piece data between storage clients and miners, both when consummating a storage deal and when retrieving the piece later.

Table of Contents

Usage

Requires go 1.13

Install the module in your package or app with go get "github.com/filecoin-project/go-data-transfer/v2"

Initialize a data transfer module

  1. Set up imports. You need, minimally, the following imports:

    package mypackage

import (
    gsimpl "github.com/ipfs/go-graphsync/impl"
    datatransfer "github.com/filecoin-project/go-data-transfer/v2/impl"
    gstransport "github.com/filecoin-project/go-data-transfer/v2/transport/graphsync"
    "github.com/libp2p/go-libp2p/core/host"
)

  2. Provide or create a libp2p host.Host

  3. You will need a transport protocol. The current default transport is graphsync. go-graphsync GraphExchange

  4. Create a data transfer by building a transport interface and then initializing a new data transfer instance

    func NewGraphsyncDataTransfer(h host.Host, gs graphsync.GraphExchange) {
    tp := gstransport.NewTransport(h.ID(), gs)
    dt := impl.NewDataTransfer(h, tp)
}

  5. If needed, build out your voucher struct and its validator.

    A push or pull request must include a voucher. The voucher's type must have been registered with the node receiving the request before it's sent, otherwise the request will be rejected.

    datatransfer.Voucher and datatransfer.Validator are the interfaces used for validation of graphsync datatransfer messages. Voucher types plus a Validator for them must be registered with the peer to whom requests will be sent.

Example Toy Voucher and Validator
type myVoucher struct {
	data string
}

func (v *myVoucher) ToBytes() ([]byte, error) {
	return []byte(v.data), nil
}

func (v *myVoucher) FromBytes(data []byte) error {
	v.data = string(data)
	return nil
}

func (v *myVoucher) Type() string {
	return "FakeDTType"
}

type myValidator struct {
	ctx                 context.Context
	ValidationsReceived chan receivedValidation
}

func (vl *myValidator) ValidatePush(
	sender peer.ID,
	voucher datatransfer.Voucher,
	baseCid cid.Cid,
	selector datamodel.Node) error {
    
    v := voucher.(*myVoucher)
    if v.data == "" || v.data != "validpush" {
        return errors.New("invalid")
    }   

	return nil
}

func (vl *myValidator) ValidatePull(
	receiver peer.ID,
	voucher datatransfer.Voucher,
	baseCid cid.Cid,
	selector datamodel.Node) error {

    v := voucher.(*myVoucher)
    if v.data == "" || v.data != "validpull" {
        return errors.New("invalid")
    }   

	return nil
}

Please see go-data-transfer/blob/master/types.go for more detail.

Register a validator

Before sending push or pull requests, you must register a datatransfer.Voucher by its reflect.Type and dataTransfer.RequestValidator for vouchers that must be sent with the request. Using the trivial examples above:

    func NewGraphsyncDatatransfer(h host.Host, gs graphsync.GraphExchange) {
        tp := gstransport.NewTransport(h.ID(), gs)
        dt := impl.NewDataTransfer(h, tp)

        vouch := &myVoucher{}
        mv := &myValidator{} 
        dt.RegisterVoucherType(reflect.TypeOf(vouch), mv)
    }

For more detail, please see the unit tests.

Open a Push or Pull Request

For a push or pull request, provide a context, a datatransfer.Voucher, a host recipient peer.ID, a baseCID cid.CID and a selector datamodel.Node. These calls return a datatransfer.ChannelID and any error:

    channelID, err := dtm.OpenPullDataChannel(ctx, recipient, voucher, baseCid, selector)
    // OR
    channelID, err := dtm.OpenPushDataChannel(ctx, recipient, voucher, baseCid, selector)
Subscribe to Events

The module allows the consumer to be notified when a graphsync Request is sent or a datatransfer push or pull request response is received:

    func ToySubscriberFunc (event Event, channelState ChannelState) {
        if event.Code == datatransfer.Error {
            // log error, flail about helplessly
            return
        }
        // 
        if channelState.Recipient() == our.PeerID && channelState.Received() > 0 {
            // log some stuff, update some state somewhere, send data to a channel, etc.
        }
    }

    dtm := SetupDataTransferManager(ctx, h, gs, baseCid, snode)
    unsubFunc := dtm.SubscribeToEvents(ToySubscriberFunc)

    // . . . later, when you don't need to know about events any more:
    unsubFunc()

Contributing

PRs are welcome! Please first read the design docs and look over the current code. PRs against master require approval of at least two maintainers. For the rest, please see our CONTRIBUTING guide.

License

This repository is dual-licensed under Apache 2.0 and MIT terms.

Copyright 2019. Protocol Labs, Inc.

Documentation

Index

Constants

View Source 
const EmptyTypeIdentifier = TypeIdentifier("")

EmptyTypeIdentifier means there is no voucher present

View Source 
const ErrAlreadySubscribed = errorType("already subscribed to events for given channel id")

ErrAlreadySubscribed indicates a subscription to events exists for the given channel

View Source 
const ErrChannelNotFound = errorType("channel not found")

ErrChannelNotFound means the channel this command was issued for does not exist

View Source 
const ErrHandlerAlreadySet = errorType("already set event handler")

ErrHandlerAlreadySet means an event handler was already set for this instance of hooks

View Source 
const ErrHandlerNotSet = errorType("event handler has not been set")

ErrHandlerNotSet means you cannot issue commands to this interface because the handler has not been set

View Source 
const ErrPause = errorType("pause channel")

ErrPause is a special error that the DataReceived / DataSent hooks can use to pause the channel

View Source 
const ErrRejected = errorType("response rejected")

ErrRejected indicates a request was not accepted

View Source 
const ErrResume = errorType("resume channel")

ErrResume is a special error that the RequestReceived / ResponseReceived hooks can use to resume the channel

View Source 
const ErrUnsupported = errorType("unsupported")

ErrUnsupported indicates an operation is not supported by the transport protocol

Variables

View Source 
var Events = map[EventCode]string{
	Open:                        "Open",
	Accept:                      "Accept",
	Restart:                     "Restart",
	DataReceived:                "DataReceived",
	DataSent:                    "DataSent",
	Cancel:                      "Cancel",
	Error:                       "Error",
	CleanupComplete:             "CleanupComplete",
	NewVoucher:                  "NewVoucher",
	NewVoucherResult:            "NewVoucherResult",
	PauseInitiator:              "PauseInitiator",
	ResumeInitiator:             "ResumeInitiator",
	PauseResponder:              "PauseResponder",
	ResumeResponder:             "ResumeResponder",
	FinishTransfer:              "FinishTransfer",
	ResponderCompletes:          "ResponderCompletes",
	ResponderBeginsFinalization: "ResponderBeginsFinalization",
	BeginFinalizing:             "BeginFinalizing",
	Disconnected:                "Disconnected",
	Complete:                    "Complete",
	CompleteCleanupOnRestart:    "CompleteCleanupOnRestart",
	DataQueued:                  "DataQueued",
	DataQueuedProgress:          "DataQueuedProgress",
	DataSentProgress:            "DataSentProgress",
	DataReceivedProgress:        "DataReceivedProgress",
	RequestTimedOut:             "RequestTimedOut",
	SendDataError:               "SendDataError",
	ReceiveDataError:            "ReceiveDataError",
	TransferRequestQueued:       "TransferRequestQueued",
	RequestCancelled:            "RequestCancelled",
	Opened:                      "Opened",
	SetDataLimit:                "SetDataLimit",
	SetRequiresFinalization:     "SetRequiresFinalization",
	DataLimitExceeded:           "DataLimitExceeded",
	TransferInitiated:           "TransferInitiated",
	SendMessageError:            "SendMessageError",
}

Events are human readable names for data transfer events

View Source 
var FinalizationStatuses = statusList{Finalizing, Completed, Completing}
View Source 
var NotAcceptedStates = statusList{
	Requested,
	AwaitingAcceptance,
	Cancelled,
	Cancelling,
	Failed,
	Failing,
	ChannelNotFoundError}
View Source 
var (
	// ProtocolDataTransfer1_2 is the protocol identifier for the latest
	// version of data-transfer (supports do-not-send-first-blocks extension)
	ProtocolDataTransfer1_2 protocol.ID = "/fil/datatransfer/1.2.0"
)
View Source 
var Statuses = map[Status]string{

	Requested:                           "Requested",
	Ongoing:                             "Ongoing",
	TransferFinished:                    "TransferFinished",
	ResponderCompleted:                  "ResponderCompleted",
	Finalizing:                          "Finalizing",
	Completing:                          "Completing",
	Completed:                           "Completed",
	Failing:                             "Failing",
	Failed:                              "Failed",
	Cancelling:                          "Cancelling",
	Cancelled:                           "Cancelled",
	InitiatorPaused:                     "InitiatorPaused",
	ResponderPaused:                     "ResponderPaused",
	BothPaused:                          "BothPaused",
	ResponderFinalizing:                 "ResponderFinalizing",
	ResponderFinalizingTransferFinished: "ResponderFinalizingTransferFinished",
	ChannelNotFoundError:                "ChannelNotFoundError",
	Queued:                              "Queued",
	AwaitingAcceptance:                  "AwaitingAcceptance",
}

Statuses are human readable names for data transfer states

View Source 
var TransferCompleteStates = statusList{
	TransferFinished,
	ResponderFinalizingTransferFinished,
	Finalizing,
	Completed,
	Completing,
	Failing,
	Failed,
	Cancelling,
	Cancelled,
	ChannelNotFoundError,
}
View Source 
var TransferringStates = statusList{
	Ongoing,
	ResponderCompleted,
	ResponderFinalizing,
	AwaitingAcceptance,
}

Functions

This section is empty.

Types

type Channel 

type Channel interface {
	// TransferID returns the transfer id for this channel
	TransferID() TransferID

	// BaseCID returns the CID that is at the root of this data transfer
	BaseCID() cid.Cid

	// Selector returns the IPLD selector for this data transfer (represented as
	// an IPLD node)
	Selector() datamodel.Node

	// Voucher returns the initial voucher for this data transfer
	Voucher() TypedVoucher

	// Sender returns the peer id for the node that is sending data
	Sender() peer.ID

	// Recipient returns the peer id for the node that is receiving data
	Recipient() peer.ID

	// TotalSize returns the total size for the data being transferred
	TotalSize() uint64

	// IsPull returns whether this is a pull request
	IsPull() bool

	// ChannelID returns the ChannelID for this request
	ChannelID() ChannelID

	// OtherPeer returns the counter party peer for this channel
	OtherPeer() peer.ID
}

Channel represents all the parameters for a single data transfer

type ChannelID 

type ChannelID struct {
	Initiator peer.ID
	Responder peer.ID
	ID        TransferID
}

ChannelID is a unique identifier for a channel, distinct by both the other party's peer ID + the transfer ID

func (*ChannelID) MarshalCBOR 

func (t *ChannelID) MarshalCBOR(w io.Writer) error

func (ChannelID) OtherParty 

func (c ChannelID) OtherParty(thisPeer peer.ID) peer.ID

OtherParty returns the peer on the other side of the request, depending on whether this peer is the initiator or responder

func (ChannelID) String 

func (c ChannelID) String() string

func (*ChannelID) UnmarshalCBOR 

func (t *ChannelID) UnmarshalCBOR(r io.Reader) (err error)

type ChannelStage 

type ChannelStage struct {
	// Human-readable fields.
	// TODO: these _will_ need to be converted to canonical representations, so
	//  they are machine readable.
	Name        string
	Description string

	// Timestamps.
	// TODO: may be worth adding an exit timestamp. It _could_ be inferred from
	//  the start of the next stage, or from the timestamp of the last log line
	//  if this is a terminal stage. But that's non-determistic and it relies on
	//  assumptions.
	CreatedTime cbg.CborTime
	UpdatedTime cbg.CborTime

	// Logs contains a detailed timeline of events that occurred inside
	// this stage.
	Logs []*Log
}

ChannelStage traces the execution of a data transfer channel stage.

EXPERIMENTAL; subject to change.

func (*ChannelStage) MarshalCBOR 

func (t *ChannelStage) MarshalCBOR(w io.Writer) error

func (*ChannelStage) UnmarshalCBOR 

func (t *ChannelStage) UnmarshalCBOR(r io.Reader) (err error)

type ChannelStages 

type ChannelStages struct {
	// Stages contains an entry for every stage the channel has gone through.
	// Each stage then contains logs.
	Stages []*ChannelStage
}

ChannelStages captures a timeline of the progress of a data transfer channel, grouped by stages.

EXPERIMENTAL; subject to change.

func (*ChannelStages) AddLog 

func (cs *ChannelStages) AddLog(stage, msg string)

AddLog adds a log to the specified stage, creating the stage if it doesn't exist yet.

EXPERIMENTAL; subject to change.

func (*ChannelStages) GetStage 

func (cs *ChannelStages) GetStage(stage string) *ChannelStage

GetStage returns the ChannelStage object for a named stage, or nil if not found.

TODO: the input should be a strongly-typed enum instead of a free-form string. TODO: drop Get from GetStage to make this code more idiomatic. Return a 

second ok boolean to make it even more idiomatic.

EXPERIMENTAL; subject to change.

func (*ChannelStages) MarshalCBOR 

func (t *ChannelStages) MarshalCBOR(w io.Writer) error

func (*ChannelStages) UnmarshalCBOR 

func (t *ChannelStages) UnmarshalCBOR(r io.Reader) (err error)

type ChannelState 

type ChannelState interface {
	Channel

	// SelfPeer returns the peer this channel belongs to
	SelfPeer() peer.ID

	// Status is the current status of this channel
	Status() Status

	// Sent returns the number of bytes sent
	Sent() uint64

	// Received returns the number of bytes received
	Received() uint64

	// Message offers additional information about the current status
	Message() string

	// Vouchers returns all vouchers sent on this channel
	Vouchers() []TypedVoucher

	// VoucherResults are results of vouchers sent on the channel
	VoucherResults() []TypedVoucher

	// LastVoucher returns the last voucher sent on the channel
	LastVoucher() TypedVoucher

	// LastVoucherResult returns the last voucher result sent on the channel
	LastVoucherResult() TypedVoucher

	// 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
	ReceivedCidsTotal() int64

	// 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
	QueuedCidsTotal() int64

	// 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
	SentCidsTotal() int64

	// Queued returns the number of bytes read from the node and queued for sending
	Queued() uint64

	// DataLimit is the maximum data that can be transferred on this channel before
	// revalidation. 0 indicates no limit.
	DataLimit() uint64

	// RequiresFinalization indicates at the end of the transfer, the channel should
	// be left open for a final settlement
	RequiresFinalization() bool

	// InitiatorPaused indicates whether the initiator of this channel is in a paused state
	InitiatorPaused() bool

	// ResponderPaused indicates whether the responder of this channel is in a paused state
	ResponderPaused() bool

	// BothPaused indicates both sides of the transfer have paused the transfer
	BothPaused() bool

	// SelfPaused indicates whether the local peer for this channel is in a paused state
	SelfPaused() bool

	// Stages returns the timeline of events this data transfer has gone through,
	// for observability purposes.
	//
	// It is unsafe for the caller to modify the return value, and changes
	// may not be persisted. It should be treated as immutable.
	Stages() *ChannelStages
}

ChannelState is channel parameters plus it's current state

type Event 

type Event struct {
	Code      EventCode // What type of event it is
	Message   string    // Any clarifying information about the event
	Timestamp time.Time // when the event happened
}

Event is a struct containing information about a data transfer event

type EventCode 

type EventCode int

EventCode is a name for an event that occurs on a data transfer channel 

const (
	// Open is an event occurs when a channel is first opened
	Open EventCode = iota

	// Accept is an event that emits when the data transfer is first accepted
	Accept

	// Restart is an event that emits when the data transfer is restarted
	Restart

	// DataReceived is emitted when data is received on the channel from a remote peer
	DataReceived

	// DataSent is emitted when data is sent on the channel to the remote peer
	DataSent

	// Cancel indicates one side has cancelled the transfer
	Cancel

	// Error is an event that emits when an error occurs in a data transfer
	Error

	// CleanupComplete emits when a request is cleaned up
	CleanupComplete

	// NewVoucher means we have a new voucher on this channel
	NewVoucher

	// NewVoucherResult means we have a new voucher result on this channel
	NewVoucherResult

	// PauseInitiator emits when the data sender pauses transfer
	PauseInitiator

	// ResumeInitiator emits when the data sender resumes transfer
	ResumeInitiator

	// PauseResponder emits when the data receiver pauses transfer
	PauseResponder

	// ResumeResponder emits when the data receiver resumes transfer
	ResumeResponder

	// FinishTransfer emits when the initiator has completed sending/receiving data
	FinishTransfer

	// ResponderCompletes emits when the initiator receives a message that the responder is finished
	ResponderCompletes

	// ResponderBeginsFinalization emits when the initiator receives a message that the responder is finilizing
	ResponderBeginsFinalization

	// BeginFinalizing emits when the responder completes its operations but awaits a response from the
	// initiator
	BeginFinalizing

	// Disconnected emits when we are not able to connect to the other party
	Disconnected

	// Complete is emitted when a data transfer is complete
	Complete

	// CompleteCleanupOnRestart is emitted when a data transfer channel is restarted to signal
	// that channels that were cleaning up should finish cleanup
	CompleteCleanupOnRestart

	// DataQueued is emitted when data is read and queued for sending to the remote peer
	DataQueued

	// DataQueuedProgress is emitted when a block is queued for sending to the
	// remote peer. It is not emitted when the block is resent.
	// It is used to measure progress of how much of the total data has been
	// queued.
	DataQueuedProgress

	// DataSentProgress is emitted when a block is sent to the remote peer.
	// It is not emitted when the block is resent.
	// It is used to measure progress of how much of the total data has
	// been sent.
	DataSentProgress

	// DataReceivedProgress is emitted the first time a block is received from
	// the remote peer. It is used to measure progress of how much of the total
	// data has been received.
	DataReceivedProgress

	// DEPRECATED in favour of RequestCancelled
	RequestTimedOut

	// SendDataError indicates that the transport layer had an error trying
	// to send data to the remote peer
	SendDataError

	// ReceiveDataError indicates that the transport layer had an error
	// receiving data from the remote peer
	ReceiveDataError

	// DEPRECATED in favor of TransferInitiated
	TransferRequestQueued

	// RequestCancelled indicates that a transport layer request was cancelled by the request opener
	RequestCancelled

	// Opened is fired when a request for data is sent from this node to a peer
	Opened

	// SetDataLimit is fired when a responder sets a limit for data it will allow
	// before pausing the request
	SetDataLimit

	// SetRequiresFinalization is fired when a responder sets a limit for data it will allow
	// before pausing the request
	SetRequiresFinalization

	// DataLimitExceeded is fired when a request exceeds it's data limit. It has the effect of
	// pausing the responder, but is distinct from PauseResponder to indicate why the pause
	// happened
	DataLimitExceeded

	// TransferInitiated indicates the transport has begun transferring data
	TransferInitiated

	// SendMessageError indicates an error sending a data transfer message
	SendMessageError
)

func (EventCode) String 

func (e EventCode) String() string

type EventsHandler 

type EventsHandler interface {
	// OnChannelOpened is called when we send a request for data to the other
	// peer on the given channel ID
	// return values are:
	// - error = ignore incoming data for this channel
	OnChannelOpened(chid ChannelID) error
	// OnResponseReceived is called when we receive a response to a request
	// - nil = continue receiving data
	// - error = cancel this request
	OnResponseReceived(chid ChannelID, msg Response) error
	// OnDataReceive is called when we receive data for the given channel ID
	// return values are:
	// - nil = proceed with sending data
	// - error = cancel this request
	// - err == ErrPause - pause this request
	OnDataReceived(chid ChannelID, link ipld.Link, size uint64, index int64, unique bool) error

	// OnDataQueued is called when data is queued for sending for the given channel ID
	// return values are:
	// - nil = proceed with sending data
	// - error = cancel this request
	// - err == ErrPause - pause this request
	OnDataQueued(chid ChannelID, link ipld.Link, size uint64, index int64, unique bool) (Message, error)

	// OnDataSent is called when we send data for the given channel ID
	OnDataSent(chid ChannelID, link ipld.Link, size uint64, index int64, unique bool) error

	// OnTransferInitiated is called when the transport layer initiates transfer
	OnTransferInitiated(chid ChannelID)

	// OnRequestReceived is called when we receive a new request to send data
	// for the given channel ID
	// return values are:
	// message = data transfer message along with reply
	// err = error
	// - nil = proceed with sending data
	// - error = cancel this request
	// - err == ErrPause - pause this request (only for new requests)
	// - err == ErrResume - resume this request (only for update requests)
	OnRequestReceived(chid ChannelID, msg Request) (Response, error)
	// OnChannelCompleted is called when we finish transferring data for the given channel ID
	// Error returns are logged but otherwise have no effect
	OnChannelCompleted(chid ChannelID, err error) error

	// OnRequestCancelled is called when a request we opened (with the given channel Id) to
	// receive data is cancelled by us.
	// Error returns are logged but otherwise have no effect
	OnRequestCancelled(chid ChannelID, err error) error

	// OnRequestDisconnected is called when a network error occurs trying to send a request
	OnRequestDisconnected(chid ChannelID, err error) error

	// OnSendDataError is called when a network error occurs sending data
	// at the transport layer
	OnSendDataError(chid ChannelID, err error) error

	// OnReceiveDataError is called when a network error occurs receiving data
	// at the transport layer
	OnReceiveDataError(chid ChannelID, err error) error

	// OnContextAugment allows the transport to attach data transfer tracing information
	// to its local context, in order to create a hierarchical trace
	OnContextAugment(chid ChannelID) func(context.Context) context.Context
}

EventsHandler are semantic data transfer events that happen as a result of graphsync hooks

type Log 

type Log struct {
	// Log is a human readable message.
	//
	// TODO: this _may_ need to be converted to a canonical data model so it
	//  is machine-readable.
	Log string

	UpdatedTime cbg.CborTime
}

Log represents a point-in-time event that occurred inside a channel stage.

EXPERIMENTAL; subject to change.

func (*Log) MarshalCBOR 

func (t *Log) MarshalCBOR(w io.Writer) error

func (*Log) UnmarshalCBOR 

func (t *Log) UnmarshalCBOR(r io.Reader) (err error)

type Manager 

type Manager interface {

	// Start initializes data transfer processing
	Start(ctx context.Context) error

	// OnReady registers a listener for when the data transfer comes on line
	OnReady(ReadyFunc)

	// Stop terminates all data transfers and ends processing
	Stop(ctx context.Context) error

	// RegisterVoucherType registers a validator for the given voucher type
	// will error if voucher type does not implement voucher
	// or if there is a voucher type registered with an identical identifier
	RegisterVoucherType(voucherType TypeIdentifier, validator RequestValidator) error

	// RegisterTransportConfigurer registers the given transport configurer to be run on requests with the given voucher
	// type
	RegisterTransportConfigurer(voucherType TypeIdentifier, configurer TransportConfigurer) error

	// open a data transfer that will send data to the recipient peer and
	// transfer parts of the piece that match the selector
	OpenPushDataChannel(ctx context.Context, to peer.ID, voucher TypedVoucher, baseCid cid.Cid, selector datamodel.Node, options ...TransferOption) (ChannelID, error)

	// open a data transfer that will request data from the sending peer and
	// transfer parts of the piece that match the selector
	OpenPullDataChannel(ctx context.Context, to peer.ID, voucher TypedVoucher, baseCid cid.Cid, selector datamodel.Node, options ...TransferOption) (ChannelID, error)

	// send an intermediate voucher as needed when the receiver sends a request for revalidation
	SendVoucher(ctx context.Context, chid ChannelID, voucher TypedVoucher) error

	// send information from the responder to update the initiator on the state of their voucher
	SendVoucherResult(ctx context.Context, chid ChannelID, voucherResult TypedVoucher) error

	// Update the validation status for a given channel, to change data limits, finalization, accepted status, and pause state
	// and send new voucher results as
	UpdateValidationStatus(ctx context.Context, chid ChannelID, validationResult ValidationResult) error

	// close an open channel (effectively a cancel)
	CloseDataTransferChannel(ctx context.Context, chid ChannelID) error

	// pause a data transfer channel (only allowed if transport supports it)
	PauseDataTransferChannel(ctx context.Context, chid ChannelID) error

	// resume a data transfer channel (only allowed if transport supports it)
	ResumeDataTransferChannel(ctx context.Context, chid ChannelID) error

	// get status of a transfer
	TransferChannelStatus(ctx context.Context, x ChannelID) Status

	// get channel state
	ChannelState(ctx context.Context, chid ChannelID) (ChannelState, error)

	// get notified when certain types of events happen
	SubscribeToEvents(subscriber Subscriber) Unsubscribe

	// get all in progress transfers
	InProgressChannels(ctx context.Context) (map[ChannelID]ChannelState, error)

	// RestartDataTransferChannel restarts an existing data transfer channel
	RestartDataTransferChannel(ctx context.Context, chid ChannelID) error
}

Manager is the core interface presented by all implementations of of the data transfer sub system

type Message 

type Message interface {
	IsRequest() bool
	IsRestart() bool
	IsNew() bool
	IsUpdate() bool
	IsPaused() bool
	IsCancel() bool
	TransferID() TransferID
	ToNet(w io.Writer) error
	ToIPLD() datamodel.Node
	MessageForProtocol(targetProtocol protocol.ID) (newMsg Message, err error)
}

Message is a message for the data transfer protocol (either request or response) that can serialize to a protobuf

type PauseableTransport 

type PauseableTransport interface {
	Transport
	// PauseChannel paused the given channel ID
	PauseChannel(ctx context.Context,
		chid ChannelID,
	) error
	// ResumeChannel resumes the given channel
	ResumeChannel(ctx context.Context,
		msg Message,
		chid ChannelID,
	) error
}

PauseableTransport is a transport that can also pause and resume channels

type ReadyFunc 

type ReadyFunc func(error)

ReadyFunc is function that gets called once when the data transfer module is ready

type Request 

type Request interface {
	Message
	IsPull() bool
	IsVoucher() bool
	VoucherType() TypeIdentifier
	Voucher() (datamodel.Node, error)
	TypedVoucher() (TypedVoucher, error)
	BaseCid() cid.Cid
	Selector() (datamodel.Node, error)
	IsRestartExistingChannelRequest() bool
	RestartChannelId() (ChannelID, error)
}

Request is a response message for the data transfer protocol

type RequestValidator 

type RequestValidator interface {
	// ValidatePush validates a push request received from the peer that will send data
	// -- All information about the validation operation is contained in ValidationResult,
	// including if it was rejected. Information about why a rejection occurred is embedded
	// in the VoucherResult.
	// -- error indicates something went wrong with the actual process of trying to validate
	ValidatePush(
		chid ChannelID,
		sender peer.ID,
		voucher datamodel.Node,
		baseCid cid.Cid,
		selector datamodel.Node) (ValidationResult, error)
	// ValidatePull validates a pull request received from the peer that will receive data
	// -- All information about the validation operation is contained in ValidationResult,
	// including if it was rejected. Information about why a rejection occurred should be embedded
	// in the VoucherResult.
	// -- error indicates something went wrong with the actual process of trying to validate
	ValidatePull(
		chid ChannelID,
		receiver peer.ID,
		voucher datamodel.Node,
		baseCid cid.Cid,
		selector datamodel.Node) (ValidationResult, error)

	// ValidateRestart validates restarting a request
	// -- All information about the validation operation is contained in ValidationResult,
	// including if it was rejected. Information about why a rejection occurred should be embedded
	// in the VoucherResult.
	// -- error indicates something went wrong with the actual process of trying to validate
	ValidateRestart(channelID ChannelID, channel ChannelState) (ValidationResult, error)
}

RequestValidator is an interface implemented by the client of the data transfer module to validate requests

type Response 

type Response interface {
	Message
	IsValidationResult() bool
	IsComplete() bool
	Accepted() bool
	VoucherResultType() TypeIdentifier
	VoucherResult() (datamodel.Node, error)
	EmptyVoucherResult() bool
}

Response is a response message for the data transfer protocol

type Status 

type Status uint64

Status is the status of transfer for a given channel 

const (
	// Requested means a data transfer was requested by has not yet been approved
	Requested Status = iota

	// Ongoing means the data transfer is in progress
	Ongoing

	// TransferFinished indicates the initiator is done sending/receiving
	// data but is awaiting confirmation from the responder
	TransferFinished

	// ResponderCompleted indicates the initiator received a message from the
	// responder that it's completed
	ResponderCompleted

	// Finalizing means the responder is awaiting a final message from the initator to
	// consider the transfer done
	Finalizing

	// Completing just means we have some final cleanup for a completed request
	Completing

	// Completed means the data transfer is completed successfully
	Completed

	// Failing just means we have some final cleanup for a failed request
	Failing

	// Failed means the data transfer failed
	Failed

	// Cancelling just means we have some final cleanup for a cancelled request
	Cancelling

	// Cancelled means the data transfer ended prematurely
	Cancelled

	// DEPRECATED: Use InitiatorPaused() method on ChannelState
	InitiatorPaused

	// DEPRECATED: Use ResponderPaused() method on ChannelState
	ResponderPaused

	// DEPRECATED: Use BothPaused() method on ChannelState
	BothPaused

	// ResponderFinalizing is a unique state where the responder is awaiting a final voucher
	ResponderFinalizing

	// ResponderFinalizingTransferFinished is a unique state where the responder is awaiting a final voucher
	// and we have received all data
	ResponderFinalizingTransferFinished

	// ChannelNotFoundError means the searched for data transfer does not exist
	ChannelNotFoundError

	// Queued indicates a data transfer request has been accepted, but is not actively transfering yet
	Queued

	// AwaitingAcceptance indicates a transfer request is actively being processed by the transport
	// even if the remote has not yet responded that it's accepted the transfer. Such a state can
	// occur, for example, in a requestor-initiated transfer that starts processing prior to receiving
	// acceptance from the server.
	AwaitingAcceptance
)

func (Status) InFinalization 

func (s Status) InFinalization() bool

func (Status) IsAccepted 

func (s Status) IsAccepted() bool

func (Status) String 

func (s Status) String() string

func (Status) TransferComplete 

func (s Status) TransferComplete() bool

func (Status) Transferring 

func (s Status) Transferring() bool

type Subscriber 

type Subscriber func(event Event, channelState ChannelState)

Subscriber is a callback that is called when events are emitted

type TransferConfig 

type TransferConfig interface {
	EventsCb() Subscriber
	TransportOptions() []TransportOption
}

TransferConfig accesses transfer properties

func FromOptions 

func FromOptions(options []TransferOption) TransferConfig

FromOptions builds a config from an options list

type TransferID 

type TransferID uint64

TransferID is an identifier for a data transfer, shared between request/responder and unique to the requester

type TransferOption 

type TransferOption func(*transferConfig)

TransferOption customizes a single transfer

func WithSubscriber 

func WithSubscriber(eventsCb Subscriber) TransferOption

WithSubscriber dispatches only events for this specific channel transfer

func WithTransportOptions 

func WithTransportOptions(transportOptions ...TransportOption) TransferOption

type Transport 

type Transport interface {
	// OpenChannel initiates an outgoing request for the other peer to send data
	// to us on this channel
	// Note: from a data transfer symantic standpoint, it doesn't matter if the
	// request is push or pull -- OpenChannel is called by the party that is
	// intending to receive data
	OpenChannel(
		ctx context.Context,
		dataSender peer.ID,
		channelID ChannelID,
		root ipld.Link,
		stor datamodel.Node,
		channel ChannelState,
		msg Message,
	) error

	// CloseChannel closes the given channel
	CloseChannel(ctx context.Context, chid ChannelID) error
	// SetEventHandler sets the handler for events on channels
	SetEventHandler(events EventsHandler) error
	// CleanupChannel is called on the otherside of a cancel - removes any associated
	// data for the channel
	CleanupChannel(chid ChannelID)
	Shutdown(ctx context.Context) error
}

Transport defines the interface for a transport layer for data transfer. Where the data transfer manager will coordinate setting up push and pull requests, validation, etc, the transport layer is responsible for moving data back and forth, and may be medium specific. For example, some transports may have the ability to pause and resume requests, while others may not. Some may support individual data events, while others may only support message events. Some transport layers may opt to use the actual data transfer network protocols directly while others may be able to encode messages in their own data protocol.

Transport is the minimum interface that must be satisfied to serve as a datatransfer transport layer. Transports must be able to open (open is always called by the receiving peer) and close channels, and set at an event handler

type TransportConfigurer 

type TransportConfigurer func(chid ChannelID, voucher TypedVoucher) []TransportOption

TransportConfigurer provides a mechanism to provide transport specific configuration for a given voucher type

type TransportOption 

type TransportOption func(chid ChannelID, transport Transport) error

type TypeIdentifier 

type TypeIdentifier string

TypeIdentifier is a unique string identifier for a type of encodable object in a registry

type TypedVoucher 

type TypedVoucher struct {
	Voucher datamodel.Node
	Type    TypeIdentifier
}

TypedVoucher is a voucher or voucher result in IPLD form and an associated type identifier for that voucher or voucher result

func (TypedVoucher) Equals 

func (tv1 TypedVoucher) Equals(tv2 TypedVoucher) bool

Equals is a utility to compare that two TypedVouchers are the same - both type and the voucher's IPLD content

type Unsubscribe 

type Unsubscribe func()

Unsubscribe is a function that gets called to unsubscribe from data transfer events

type ValidationResult 

type ValidationResult struct {
	// Accepted indicates whether the request was accepted. If a request is not
	// accepted, the request fails. This is true for revalidation as well
	Accepted bool
	// VoucherResult provides information to the other party about what happened
	// with the voucher
	VoucherResult *TypedVoucher
	// ForcePause indicates whether the request should be paused, regardless
	// of data limit and finalization status
	ForcePause bool
	// DataLimit specifies how much data this voucher is good for. When the amount
	// of data specified is reached (or shortly after), the request will pause
	// pending revalidation. 0 indicates no limit.
	DataLimit uint64
	// RequiresFinalization indicates at the end of the transfer, the channel should
	// be left open for a final settlement
	RequiresFinalization bool
}

ValidationResult describes the result of validating a voucher

func (ValidationResult) Equals 

func (vr ValidationResult) Equals(vr2 ValidationResult) bool

Equals checks the deep equality of two ValidationResult values

func (ValidationResult) LeaveRequestPaused 

func (vr ValidationResult) LeaveRequestPaused(chst ChannelState) bool

LeaveRequestPaused indicates whether all conditions are met to resume a request

Source Files

View all Source files

Directories

Path Synopsis
benchmarks
testinstance
testnet
internal
internal/migrations
message1_1prime
types
transport
graphsync
graphsync/extension
graphsync/testharness

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.