Skip to main content

utils

Helper module for federated transport.

Classes​

MessageServiceStubWrapper​

class MessageServiceStubWrapper(config: MessageServiceConfig):

Wrapper around the MessageServiceStub to allow channel updates.

We need this to provide an indirection to the underlying MessageServiceStub that doesn't allow the channel to be mutated.

Constructor.

Arguments

  • channel: A grpc.Channel.

Ancestors​

Methods​

AcknowledgeMessage​

async def AcknowledgeMessage(    self,    data: Acknowledgement,    /,    *,    metadata: Sequence[tuple[str, Any]],    timeout: float | None = 20.0,) ‑> SuccessResponse:

AcknowledgeMessage GRPC Endpoint.

Diagnostics​

async def Diagnostics(    self,    data: DiagnosticsParameters,    /,    *,    metadata: Sequence[tuple[str, Any]],    timeout: float | None = 20.0,) ‑> ServiceDiagnostics:

Diagnostics GRPC Endpoint.

GetBitfountMessage​

async def GetBitfountMessage(    self,    data: CommunicationDetails,    /,    *,    metadata: Sequence[tuple[str, Any]],    timeout: float | None = 20.0,) ‑> BitfountMessage:

GetBitfountMessage GRPC Endpoint.

GetLargeObjectStorage​

async def GetLargeObjectStorage(    self,    data: LargeStorageRequest,    /,    *,    metadata: Sequence[tuple[str, Any]],    timeout: float | None = 20.0,) ‑> BlobStorageData:

GetLargeObjectStorage GRPC Endpoint.

InitiateTask​

async def InitiateTask(    self,    data: BitfountTasks,    /,    *,    metadata: Sequence[tuple[str, Any]],    timeout: float | None = 20.0,) ‑> CommunicationDetails:

InitiateTask GRPC Endpoint.

PodConnect​

async def PodConnect(    self,    data: PodData,    /,    *,    metadata: Sequence[tuple[str, Any]],    timeout: float | None = 20.0,) ‑> SuccessResponse:

PodConnect GRPC Endpoint.

SendBitfountMessage​

async def SendBitfountMessage(    self,    data: BitfountMessage,    /,    *,    metadata: Sequence[tuple[str, Any]],    timeout: float | None = 20.0,) ‑> SuccessResponse:

SendBitfountMessage GRPC Endpoint.

SetupTask​

async def SetupTask(    self,    data: TaskTransferRequests,    /,    *,    metadata: Sequence[tuple[str, Any]],    timeout: float | None = 20.0,) ‑> TaskTransferMetadata:

SetupTask GRPC Endpoint.

atomic_get_and_ack​

async def atomic_get_and_ack(    self,    metadata: list[tuple[str, str]] | Callable[[], Awaitable[list[tuple[str, str]]]],    mailbox_id: str | None = None,    compute_cluster_id: str | None = None,    timeout: float | None = None,) ‑> BitfountMessage:

Get and acknowledge gRPC message in one atomic operation.

Will either return the retrieved and ACKed message or raise an RpcError if something goes wrong at either stage or if there is no message available.

Arguments

  • metadata: gRPC call metadata, or an async zero-argument callable returning it. _auto_retry_grpc re-invokes this method for every retry attempt, so a provider is awaited afresh each time; a plain list is fixed for the lifetime of the retry loop. Long-lived callers whose credential can expire mid-loop must pass a provider, and it is that provider's job not to block the event loop while resolving.
  • mailbox_id: Mailbox to read from. Defaults to the caller's own.
  • compute_cluster_id: Compute cluster the mailbox belongs to.
  • timeout: Per-call gRPC timeout in seconds.

Returns The retrieved and acknowledged message.

get_channel_state​

def get_channel_state(self) ‑> grpc.ChannelConnectivity:

Get the state of the underlying GRPC channel.

reset_channel​

def reset_channel(self) ‑> None:

Reset the underlying GRPC channel.