bprotocol

Documentation

Index

• Constants
• type CallbackHandler
  • func NewCallbackHandler(params CallbackHandlerParams) *CallbackHandler
• type CallbackHandlerParams
• type CallbackProxy
  • func NewCallbackProxy(params CallbackProxyParams) *CallbackProxy
  • func (p *CallbackProxy) OnBidComplete(ctx context.Context, result compute.BidResult)
  • func (p *CallbackProxy) OnCancelComplete(ctx context.Context, result compute.CancelResult)
  • func (p *CallbackProxy) OnComputeFailure(ctx context.Context, result compute.ComputeError)
  • func (p *CallbackProxy) OnRunComplete(ctx context.Context, result compute.RunResult)
  • func (p *CallbackProxy) RegisterLocalComputeCallback(callback compute.Callback)
• type CallbackProxyParams
• type ComputeHandler
  • func NewComputeHandler(params ComputeHandlerParams) *ComputeHandler
• type ComputeHandlerParams
• type ComputeProxy
  • func NewComputeProxy(params ComputeProxyParams) *ComputeProxy
  • func (p *ComputeProxy) AskForBid(ctx context.Context, request compute.AskForBidRequest) (compute.AskForBidResponse, error)
  • func (p *ComputeProxy) BidAccepted(ctx context.Context, request compute.BidAcceptedRequest) (compute.BidAcceptedResponse, error)
  • func (p *ComputeProxy) BidRejected(ctx context.Context, request compute.BidRejectedRequest) (compute.BidRejectedResponse, error)
  • func (p *ComputeProxy) CancelExecution(ctx context.Context, request compute.CancelExecutionRequest) (compute.CancelExecutionResponse, error)
  • func (p *ComputeProxy) ExecutionLogs(ctx context.Context, request compute.ExecutionLogsRequest) (<-chan *concurrency.AsyncResult[models.ExecutionLog], error)
  • func (p *ComputeProxy) RegisterLocalComputeEndpoint(endpoint compute.Endpoint)
• type ComputeProxyParams
• type Result
  • func (r *Result[T]) Rehydrate() (T, error)

Constants

const (
	ComputeServiceName    = "bacalhau.compute"
	AskForBidProtocolID   = "/bacalhau/compute/ask_for_bid/1.0.0"
	BidAcceptedProtocolID = "/bacalhau/compute/bid_accepted/1.0.0"
	BidRejectedProtocolID = "/bacalhau/compute/bid_rejected/1.0.0"
	CancelProtocolID      = "/bacalhau/compute/cancel/1.0.0"
	ExecutionLogsID       = "/bacalhau/compute/execution_logs/1.0.0"

	CallbackServiceName = "bacalhau.callback"
	OnBidComplete       = "/bacalhau/callback/on_bid_complete/1.0.0"
	OnRunComplete       = "/bacalhau/callback/on_run_complete/1.0.0"
	OnCancelComplete    = "/bacalhau/callback/on_cancel_complete/1.0.0"
	OnComputeFailure    = "/bacalhau/callback/on_compute_failure/1.0.0"
)

Types

type CallbackHandler

type CallbackHandler struct {
	// contains filtered or unexported fields
}

CallbackHandler is a handler for callback events that registers for incoming libp2p requests to Bacalhau callback protocol, and delegates the handling of the request to the provided callback.

func NewCallbackHandler

func NewCallbackHandler(params CallbackHandlerParams) *CallbackHandler

type CallbackHandlerParams

type CallbackHandlerParams struct {
	Host     host.Host
	Callback compute.Callback
}

type CallbackProxy

type CallbackProxy struct {
	// contains filtered or unexported fields
}

CallbackProxy is a proxy for a compute.Callback that can be used to send compute callbacks to the requester node, such as when the execution is completed or when a failure occurs. The proxy can forward callbacks to a remote requester node, or locally if the node is the requester and a LocalCallback is provided.

func NewCallbackProxy

func NewCallbackProxy(params CallbackProxyParams) *CallbackProxy

func (*CallbackProxy) OnBidComplete

func (p *CallbackProxy) OnBidComplete(ctx context.Context, result compute.BidResult)

func (*CallbackProxy) OnCancelComplete

func (p *CallbackProxy) OnCancelComplete(ctx context.Context, result compute.CancelResult)

func (*CallbackProxy) OnComputeFailure

func (p *CallbackProxy) OnComputeFailure(ctx context.Context, result compute.ComputeError)

func (*CallbackProxy) OnRunComplete

func (p *CallbackProxy) OnRunComplete(ctx context.Context, result compute.RunResult)

func (*CallbackProxy) RegisterLocalComputeCallback

func (p *CallbackProxy) RegisterLocalComputeCallback(callback compute.Callback)

type CallbackProxyParams

type CallbackProxyParams struct {
	Host          host.Host
	LocalCallback compute.Callback
}

type ComputeHandler

type ComputeHandler struct {
	// contains filtered or unexported fields
}

ComputeHandler is a handler for compute requests that registers for incoming libp2p requests to Bacalhau compute protocol, and delegates the requests to the compute endpoint.

func NewComputeHandler

func NewComputeHandler(params ComputeHandlerParams) *ComputeHandler

type ComputeHandlerParams

type ComputeHandlerParams struct {
	Host            host.Host
	ComputeEndpoint compute.Endpoint
}

type ComputeProxy

type ComputeProxy struct {
	// contains filtered or unexported fields
}

ComputeProxy is a proxy to a compute node endpoint that will forward requests to remote compute nodes, or to a local compute node if the target peer ID is the same as the local host, and a LocalEndpoint implementation is provided.

func NewComputeProxy

func NewComputeProxy(params ComputeProxyParams) *ComputeProxy

func (*ComputeProxy) AskForBid

func (p *ComputeProxy) AskForBid(ctx context.Context, request compute.AskForBidRequest) (compute.AskForBidResponse, error)

func (*ComputeProxy) BidAccepted

func (p *ComputeProxy) BidAccepted(ctx context.Context, request compute.BidAcceptedRequest) (compute.BidAcceptedResponse, error)

func (*ComputeProxy) BidRejected

func (p *ComputeProxy) BidRejected(ctx context.Context, request compute.BidRejectedRequest) (compute.BidRejectedResponse, error)

func (*ComputeProxy) CancelExecution

func (p *ComputeProxy) CancelExecution(
	ctx context.Context, request compute.CancelExecutionRequest) (compute.CancelExecutionResponse, error)

func (*ComputeProxy) ExecutionLogs

func (p *ComputeProxy) ExecutionLogs(ctx context.Context, request compute.ExecutionLogsRequest) (
	<-chan *concurrency.AsyncResult[models.ExecutionLog], error)

func (*ComputeProxy) RegisterLocalComputeEndpoint

func (p *ComputeProxy) RegisterLocalComputeEndpoint(endpoint compute.Endpoint)

type ComputeProxyParams

type ComputeProxyParams struct {
	Host          host.Host
	LocalEndpoint compute.Endpoint // optional in case this host is also a compute node and to allow local calls
}

type Result

type Result[T any] struct {
	Response T
	Error    string
}

func (*Result[T]) Rehydrate

func (r *Result[T]) Rehydrate() (T, error)