///|
/// Fatal runtime failures reported by Discord.
pub(all) suberror BotError {
/// Close codes 4013 and 4014 indicate invalid or disallowed gateway
/// intents. Check explicit intents and privileged-intent approval.
FatallyClosed(code~ : Int)
/// A shard id/count selection cannot be represented by Discord's sharding
/// rules.
InvalidShardConfig(message~ : String)
/// The identify session allowance returned by `GET /gateway/bot` cannot
/// cover the next IDENTIFY. At startup `required` counts the selected
/// shards without a saved session (a saved session sends RESUME first);
/// later it is 1 for the shard that was about to identify.
SessionStartLimitExceeded(
required~ : Int,
remaining~ : Int,
reset_after_ms~ : Int64
)
} derive(Debug)
///|
pub extend BotError with Show::{to_string}
///|
/// Human-readable failure message, including remediation hints for the
/// intent-related close codes.
pub impl Show for BotError with fn output(self, logger) {
match self {
FatallyClosed(code~) => {
logger.write_string("gateway fatally closed with code \{code}")
if code == 4013 || code == 4014 {
logger.write_string(
"; verify configured gateway intents and privileged-intent approval",
)
}
}
InvalidShardConfig(message~) =>
logger.write_string("invalid shard configuration: \{message}")
SessionStartLimitExceeded(required~, remaining~, reset_after_ms~) =>
logger.write_string(
"gateway session start limit exceeded: need \{required}, have \{remaining}; resets in \{reset_after_ms}ms",
)
}
}
///|
priv struct TypedEventHandler {
kind : @model.EventKind
required_intents : @model.Intents
dispatch : async (GatewayCtx, @model.Event) -> Unit
}
///|
priv struct ServiceHandler {
name : String
dispatch : async (GatewayCtx) -> Unit
}
///|
/// Gateway executor for a gateway-free application core on native, JavaScript,
/// and Wasm. Voice connections are native-only.
pub struct Bot {
priv app_ : @app.App
priv token_ : String
priv intents_ : @model.Intents?
priv capabilities_ : @model.GatewayCapabilities
priv client_ : @dhttp.Client?
priv gateway_url_ : String?
priv shards_ : ShardConfig
priv compress_ : Bool
priv resume_ : Map[Int, BotSession]
priv starting_ : Ref[Bool]
priv running_shards_ : Ref[Array[ManagedShard]]
priv identify_queue_ : &@queue.IdentifyQueue?
priv sync_ : @app.CommandScope?
priv sync_unowned_ : @framework.UnownedCommands
priv typed_events_ : Array[TypedEventHandler]
priv raw_events_ : Array[async (GatewayCtx, @model.Event) -> Unit]
priv event_middleware_ : Array[EventMiddleware]
priv mut cache_ : @cache.InMemoryCache?
priv decode_errors_ : Array[(String, Json) -> Unit]
priv telemetry_ : Array[(@telemetry.TelemetryEvent) -> Unit raise]
priv services_ : Array[ServiceHandler]
priv collector_ : EventCollector
// Production uses Shard::start's WebSocket connector. Whitebox tests inject
// the same in-memory GatewayTransport seam used by gateway_test.mbt.
priv connector_ : Ref[(async (String) -> &@gateway.GatewayTransport)?]
// Whitebox seams for `/gateway/bot` and Discord's five-second identify
// interval. Production always uses the REST response and 5250ms margin.
priv gateway_bot_info_ : Ref[@model.GatewayBotInfo?]
priv identify_spacing_ms_ : Ref[Int]
}
///|
/// Create a gateway runner for `app`. When `intents` is omitted, the bot
/// infers the union of intents required by its typed event subscriptions at
/// `run` time (privileged intents are only inferred, never invented — they
/// must still be enabled in the developer portal). `client` supplies a
/// preconfigured REST client; by default one is built from `token`.
/// `shards`, `compress`, and `identify_queue` configure sharding, transport
/// compression, and cross-process identify coordination. Every IDENTIFY, at
/// startup or after a lost session, first reads `session_start_limit` from
/// `GET /gateway/bot`; an exhausted limit ends `run` with
/// `SessionStartLimitExceeded` instead of risking a token reset. `capabilities`
/// is sent in Identify (see `@model.GatewayCapabilities`).
///
/// When `sync` is omitted, the bot performs no command synchronization. When
/// set, synchronization runs once per process after the first READY. It uses
/// Discord's bulk overwrite and deletes commands in that scope that this App
/// does not declare, except for entry points, which are always preserved.
/// `sync_unowned=Keep` also preserves commands registered by other owners.
/// Deletions are reported through the app's warning hook. In multi-process
/// sharding, enable synchronization in exactly one process.
/// `resume` seeds saved Bot sessions per shard id so those shards RESUME instead
/// of identifying (see `Bot::sessions`). It restores gateway identity, not
/// optional cache contents or application-owned state. `run` rejects a saved
/// session whose gateway id disagrees with its READY session id.
pub fn Bot::Bot(
app : @app.App,
token~ : String,
intents? : @model.Intents,
capabilities? : @model.GatewayCapabilities = @model.GatewayCapabilities::none(),
client? : @dhttp.Client,
gateway_url? : String,
shards? : ShardConfig = Single(id=0, count=1),
compress? : Bool = false,
identify_queue? : &@queue.IdentifyQueue,
sync? : @app.CommandScope,
sync_unowned? : @framework.UnownedCommands = Delete,
resume? : Map[Int, BotSession] = Map([]),
) -> Bot {
{
app_: app,
token_: token,
intents_: intents,
capabilities_: capabilities,
client_: client,
gateway_url_: gateway_url,
shards_: shards,
compress_: compress,
resume_: {
let snapshots : Map[Int, BotSession] = Map([])
for shard_id, snapshot in resume {
snapshots[shard_id] = clone_bot_session(snapshot)
}
snapshots
},
starting_: Ref(false),
running_shards_: Ref([]),
identify_queue_: identify_queue,
sync_: sync,
sync_unowned_: sync_unowned,
typed_events_: [],
raw_events_: [],
event_middleware_: [],
cache_: None,
decode_errors_: [],
telemetry_: [],
services_: [],
collector_: EventCollector(),
connector_: Ref(None),
gateway_bot_info_: Ref(None),
identify_spacing_ms_: Ref(5250),
}
}
///|
/// Current resumable gateway state paired with READY metadata, keyed by shard
/// id. A supplied snapshot is available from the start of `run` until its
/// shard replaces, drops, or invalidates it, including while `run` still
/// resolves the shard layout over REST; an unsaved shard appears after READY.
/// Empty while the bot is not running. Returned values do not alias mutable
/// runtime state.
pub fn Bot::sessions(self : Bot) -> Map[Int, BotSession] {
let sessions : Map[Int, BotSession] = Map([])
if self.starting_.val {
for shard_id, saved in self.resume_ {
sessions[shard_id] = clone_bot_session(saved)
}
return sessions
}
for owner in self.running_shards_.val {
if owner.shard.session() is Some(gateway) &&
owner.context is Some(ctx) &&
gateway.id == ctx.ready_.session_id {
sessions[owner.id] = { gateway, ready: clone_ready(ctx.ready_), }
}
}
sessions
}
///|
/// Observe structured gateway, dispatch, and REST telemetry aggregated by
/// this bot. Hooks run synchronously and should return promptly.
pub fn Bot::on_telemetry(
self : Bot,
hook : (@telemetry.TelemetryEvent) -> Unit raise,
) -> Unit {
self.telemetry_.push(hook)
}
///|
fn Bot::emit_telemetry(self : Bot, event : @telemetry.TelemetryEvent) -> Unit {
for hook in self.telemetry_ {
hook(event) catch {
error => self.app_.warn("telemetry hook failed: \{Repr(error)}")
}
}
}
///|
/// Feeds every decoded gateway event into an opt-in cache before dispatching
/// typed and raw event handlers.
pub fn Bot::attach_cache(self : Bot, cache : @cache.InMemoryCache) -> Unit {
self.cache_ = Some(cache)
}
///|
/// Subscribe a typed handler to one gateway event (see `Events` for the
/// descriptors). The event's required intents join the inferred intent set
/// when `Bot(...)` omits explicit `intents`. Handlers run through the event
/// middleware chain; a raised error is reported via the app's failure policy
/// without stopping the bot.
pub fn[T] Bot::on(
self : Bot,
event : EventType[T],
handler : async (GatewayCtx, T) -> Unit,
) -> Unit {
self.typed_events_.push({
kind: event.kind(),
required_intents: event.required_intents(),
dispatch: (ctx, raw) => {
if event.project(raw) is Some(value) {
handler(ctx, value)
}
},
})
}
///|
/// Subscribe to the raw event stream: every decoded gateway dispatch,
/// including `Resumed` and `Unknown`, without payload projection. Unlike
/// `Bot::on`, this adds nothing to the inferred intent set.
pub fn Bot::on_event(
self : Bot,
handler : async (GatewayCtx, @model.Event) -> Unit,
) -> Unit {
self.raw_events_.push(handler)
}
///|
/// Observe gateway events whose known payload failed typed decoding.
///
/// `marker` is the full synthetic event type
/// `DECODE_ERROR::`, and `payload` is Discord's raw
/// event data. Handlers run synchronously in the dispatch loop and should
/// return promptly. Registering a handler does not add gateway intents or
/// widen the event filter.
pub fn Bot::on_decode_error(
self : Bot,
handler : (String, Json) -> Unit,
) -> Unit {
self.decode_errors_.push(handler)
}
///|
/// Register a long-running background task started alongside the shards when
/// the bot runs (e.g. a periodic status updater). `name` labels the service
/// in failure reports. A raised error is reported via the app's failure
/// policy and stops only that service, not the bot.
pub fn Bot::service(
self : Bot,
name~ : String,
handler : async (GatewayCtx) -> Unit,
) -> Unit {
self.services_.push({ name, dispatch: handler, })
}
///|
fn Bot::observe_decode_error(self : Bot, event : @model.Event) -> Unit {
match event {
Unknown(t~, d~) if t.has_prefix("DECODE_ERROR:") => {
// Never include `d`: payloads can contain user content and credentials.
self.app_.warn("gateway event decode failure: \{t}")
self.emit_telemetry(DecodeError(marker=t))
for handler in self.decode_errors_ {
handler(t, d)
}
}
_ => ()
}
}
///|
fn privileged_intents() -> @model.Intents {
@model.Intents::guild_members() |
@model.Intents::guild_presences() |
@model.Intents::message_content()
}
///|
/// Resolve gateway intents from typed subscriptions. `required_intents` is an
/// event-delivery union, so an explicit configuration satisfies a descriptor
/// when it intersects that union. Automatic mode follows the requested union
/// literally, then removes privileged bits; enabling those always requires an
/// explicit opt-in by the application.
fn Bot::resolved_intents(self : Bot) -> @model.Intents {
match self.intents_ {
Some(explicit) => {
for subscription in self.typed_events_ {
let required = subscription.required_intents
if required.bits() != 0U && (required & explicit).bits() == 0U {
self.app_.warn(
"event \{Repr(subscription.kind)} may not be delivered; add one of its required intents explicitly",
)
}
}
explicit
}
None => {
let mut derived = @model.Intents::none()
let privileged = privileged_intents()
for subscription in self.typed_events_ {
let required = subscription.required_intents
if (required & privileged).bits() != 0U {
self.app_.warn(
"event \{Repr(subscription.kind)} requires a privileged intent; pass intents explicitly",
)
}
derived = derived | required
}
if !self.raw_events_.is_empty() {
self.app_.warn(
"raw event handlers do not imply gateway intents; pass intents explicitly for the events they consume",
)
}
let privileged_bits = derived.bits() & privileged.bits()
@model.Intents::from_bits(derived.bits() ^ privileged_bits)
}
}
}
///|
fn event_kind(event : @model.Event) -> @model.EventKind {
match event {
ApplicationCommandPermissionsUpdate(_) =>
KApplicationCommandPermissionsUpdate
AutoModerationActionExecution(_) => KAutoModerationActionExecution
AutoModerationRuleCreate(_) => KAutoModerationRuleCreate
AutoModerationRuleDelete(_) => KAutoModerationRuleDelete
AutoModerationRuleUpdate(_) => KAutoModerationRuleUpdate
ChannelCreate(_) => KChannelCreate
ChannelDelete(_) => KChannelDelete
ChannelInfo(_) => KChannelInfo
ChannelPinsUpdate(_) => KChannelPinsUpdate
ChannelUpdate(_) => KChannelUpdate
EntitlementCreate(_) => KEntitlementCreate
EntitlementDelete(_) => KEntitlementDelete
EntitlementUpdate(_) => KEntitlementUpdate
GuildAuditLogEntryCreate(_) => KGuildAuditLogEntryCreate
GuildBanAdd(_) => KGuildBanAdd
GuildBanRemove(_) => KGuildBanRemove
GuildCreate(_) => KGuildCreate
GuildDelete(_) => KGuildDelete
GuildEmojisUpdate(_) => KGuildEmojisUpdate
GuildIntegrationsUpdate(_) => KGuildIntegrationsUpdate
GuildMemberAdd(_) => KGuildMemberAdd
GuildMemberRemove(_) => KGuildMemberRemove
GuildMembersChunk(_) => KGuildMembersChunk
GuildMemberUpdate(_) => KGuildMemberUpdate
GuildRoleCreate(_) => KGuildRoleCreate
GuildRoleDelete(_) => KGuildRoleDelete
GuildRoleUpdate(_) => KGuildRoleUpdate
GuildScheduledEventCreate(_) => KGuildScheduledEventCreate
GuildScheduledEventDelete(_) => KGuildScheduledEventDelete
GuildScheduledEventUpdate(_) => KGuildScheduledEventUpdate
GuildScheduledEventUserAdd(_) => KGuildScheduledEventUserAdd
GuildScheduledEventUserRemove(_) => KGuildScheduledEventUserRemove
GuildSoundboardSoundCreate(_) => KGuildSoundboardSoundCreate
GuildSoundboardSoundDelete(_) => KGuildSoundboardSoundDelete
GuildSoundboardSoundsUpdate(_) => KGuildSoundboardSoundsUpdate
GuildSoundboardSoundUpdate(_) => KGuildSoundboardSoundUpdate
SoundboardSounds(_) => KSoundboardSounds
GuildStickersUpdate(_) => KGuildStickersUpdate
GuildUpdate(_) => KGuildUpdate
IntegrationCreate(_) => KIntegrationCreate
IntegrationDelete(_) => KIntegrationDelete
IntegrationUpdate(_) => KIntegrationUpdate
InteractionCreate(_) => KInteractionCreate
InviteCreate(_) => KInviteCreate
InviteDelete(_) => KInviteDelete
MessageCreate(_) => KMessageCreate
MessageDelete(_) => KMessageDelete
MessageDeleteBulk(_) => KMessageDeleteBulk
MessagePollVoteAdd(_) => KMessagePollVoteAdd
MessagePollVoteRemove(_) => KMessagePollVoteRemove
MessageReactionAdd(_) => KMessageReactionAdd
MessageReactionRemove(_) => KMessageReactionRemove
MessageReactionRemoveAll(_) => KMessageReactionRemoveAll
MessageReactionRemoveEmoji(_) => KMessageReactionRemoveEmoji
MessageUpdate(_) => KMessageUpdate
PresenceUpdate(_) => KPresenceUpdate
RateLimited(_) => KRateLimited
Ready(_) => KReady
Resumed => KResumed
StageInstanceCreate(_) => KStageInstanceCreate
StageInstanceDelete(_) => KStageInstanceDelete
StageInstanceUpdate(_) => KStageInstanceUpdate
SubscriptionCreate(_) => KSubscriptionCreate
SubscriptionDelete(_) => KSubscriptionDelete
SubscriptionUpdate(_) => KSubscriptionUpdate
ThreadCreate(_) => KThreadCreate
ThreadDelete(_) => KThreadDelete
ThreadListSync(_) => KThreadListSync
ThreadMembersUpdate(_) => KThreadMembersUpdate
ThreadMemberUpdate(_) => KThreadMemberUpdate
ThreadUpdate(_) => KThreadUpdate
TypingStart(_) => KTypingStart
UserUpdate(_) => KUserUpdate
VoiceChannelEffectSend(_) => KVoiceChannelEffectSend
VoiceChannelStartTimeUpdate(_) => KVoiceChannelStartTimeUpdate
VoiceChannelStatusUpdate(_) => KVoiceChannelStatusUpdate
VoiceServerUpdate(_) => KVoiceServerUpdate
VoiceStateUpdate(_) => KVoiceStateUpdate
WebhooksUpdate(_) => KWebhooksUpdate
Unknown(..) => KUnknown
}
}
///|
fn Bot::event_filter(self : Bot) -> (@model.EventKind) -> Bool {
if self.cache_ is Some(_) || !self.raw_events_.is_empty() {
return _ => true
}
kind => {
if self.collector_.waits_for(kind) {
return true
}
if kind == KReady || kind == KInteractionCreate {
return true
}
for subscription in self.typed_events_ {
if subscription.kind == kind {
return true
}
}
false
}
}
///|
fn Bot::spawn_services(
self : Bot,
group : @async.TaskGroup[Unit],
gateway : GatewayCtx,
) -> Unit {
for service in self.services_ {
group.spawn_bg(no_wait=true, allow_failure=true, () => {
(service.dispatch)(gateway) catch {
error if @async.is_being_cancelled() => raise error
error => self.app_.report_failure(Service(name=service.name), error)
}
})
}
}
///|
async fn Bot::spawn_event_handlers(
self : Bot,
spawner : @app.Spawner,
gateway : GatewayCtx,
event : @model.Event,
) -> Unit {
let kind = event_kind(event)
let mut handler_count = self.raw_events_.length()
for subscription in self.typed_events_ {
if subscription.kind == kind {
handler_count += 1
}
}
self.emit_telemetry(EventDispatched(kind="\{Repr(kind)}", handler_count~))
for subscription in self.typed_events_ {
if subscription.kind == kind {
spawner(() => {
(subscription.dispatch)(gateway, event) catch {
error if @async.is_being_cancelled() => raise error
error => self.app_.report_failure(Event(kind~), error)
}
})
}
}
for handler in self.raw_events_ {
spawner(() => {
handler(gateway, event) catch {
error if @async.is_being_cancelled() => raise error
error => self.app_.report_failure(Event(kind~), error)
}
})
}
}
///|
async fn Bot::spawn_interaction(
self : Bot,
spawner : @app.Spawner,
framework : @framework.Framework,
interaction : @model.Interaction,
) -> Unit {
spawner(() => {
ignore(
framework.process(interaction) catch {
error if @async.is_being_cancelled() => raise error
error => {
self.app_.warn("interaction processing failed: \{Repr(error)}")
false
}
},
)
})
}
///|
/// Connect and process gateway events until graceful shutdown or a fatal close.
pub async fn Bot::run(self : Bot) -> Unit {
self.app_.validate()
if self.token_.is_empty() {
raise @app.AppConfigError::EmptyToken
}
if self.compress_ && !@gateway.zlib_stream_supported() {
raise BotError::InvalidShardConfig(
message="compress=true requires zlib-stream inflate, which this backend does not provide",
)
}
for shard_id, saved in self.resume_ {
if saved.gateway.id != saved.ready.session_id {
raise BotError::InvalidShardConfig(
message="saved session for shard \{shard_id} has mismatched gateway and READY session ids",
)
}
}
let intents = self.resolved_intents()
let owns_client = self.client_ is None
let client = self.client_.unwrap_or_else(() => Client(self.token_))
defer (if owns_client { client.close() })
client.on_telemetry(event => self.emit_telemetry(event))
// Supplied snapshots stay visible through `sessions()` until the shards
// exist, so a periodic persister never observes an empty map mid-startup.
self.starting_.val = true
defer {
self.starting_.val = false
}
let resolved = self.resolve_shards(client)
@async.with_task_group((group : @async.TaskGroup[Unit]) => {
let filter = self.event_filter()
let shard_events : @aqueue.Queue[ManagedShardEvent] = Queue(kind=Unbounded)
let identify_queue : &@queue.IdentifyQueue = match self.identify_queue_ {
Some(queue) => queue
None =>
@queue.InMemoryQueue(
max_concurrency=resolved.max_concurrency,
spacing_ms=self.identify_spacing_ms_.val,
)
}
let identify_queue : &@queue.IdentifyQueue = (
{ bot: self, client, inner: identify_queue, refusals: shard_events, } :
IdentifyGuard)
let managed_shards : Array[ManagedShard] = []
let shard_handles : Array[@gateway.Shard] = []
for shard_id in resolved.ids {
let shard = self.start_managed_shard(
group,
token=self.token_,
intents~,
gateway_url=resolved.gateway_url,
shard_id~,
shard_count=resolved.count,
identify_queue~,
event_filter=filter,
resume?=resolved.resume.get(shard_id).map(snapshot => snapshot.gateway),
)
managed_shards.push({ id: shard_id, shard, context: None, })
shard_handles.push(shard)
}
self.running_shards_.val = managed_shards
self.starting_.val = false
group.add_defer(() => self.running_shards_.val = [])
group.add_defer(() => {
@async.protect_from_cancel(() => close_shards(shard_handles))
})
for owner in managed_shards {
group.spawn_bg(no_wait=true, () => {
for ;; {
shard_events.put(Shard(owner~, event=owner.shard.next()))
}
})
}
let spawner = self.app_.spawner(group)
let shutdown_requested = Ref(false)
let mut app_ctx : @app.AppCtx? = None
let mut framework : @framework.Framework? = None
let initialize_context = (owner : ManagedShard, ready : @model.Ready) => {
if framework is None {
let router = @framework.Framework(client, ready.application.id)
let waiter : @app.ComponentWaiter = (custom_id, user, timeout_ms) => {
router.wait_for_component(custom_id~, user?, timeout_ms?)
}
let latency_ms = () => {
let mut total = 0L
let mut count = 0L
for shard in shard_handles {
if shard.latency_ms() is Some(latency) {
total += latency
count += 1L
}
}
if count == 0L {
None
} else {
Some(total / count)
}
}
let attached = self.app_.attach(
router,
client~,
application_id=ready.application.id,
waiter~,
latency_ms~,
)
app_ctx = Some(attached)
framework = Some(router)
}
guard app_ctx is Some(attached) && framework is Some(router) else {
abort("gateway runtime initialized without an application context")
}
match owner.context {
Some(existing) => {
existing.ready_ = ready
existing
}
None => {
let created = GatewayCtx::{
app_: attached,
ready_: ready,
intents_: intents,
shard_: owner.shard,
shards_: shard_handles,
cache_: self.cache_,
framework_: router,
collector_: self.collector_,
group_: group,
voice_: VoiceSessions::new(),
shutdown_requested_: shutdown_requested,
}
owner.context = Some(created)
created
}
}
}
// Prepare restored contexts before reading any replayed Gateway dispatch.
for owner in managed_shards {
if resolved.resume.get(owner.id) is Some(saved) {
initialize_context(owner, clone_ready(saved.ready)) |> ignore
}
}
let mut services_started = false
let mut sync_done = false
let start_services = (ctx : GatewayCtx) => {
if !services_started {
services_started = true
self.spawn_services(group, ctx)
}
}
for ;; {
let (owner, event) = match shard_events.get() {
Shard(owner~, event~) => (owner, event)
IdentifyRefused(error) => {
close_shards(shard_handles)
raise error
}
}
match event {
Dispatch(Ready(ready) as event) => {
if self.cache_ is Some(cache) {
cache.update(event)
}
let ctx = initialize_context(owner, ready)
if !sync_done {
sync_done = true
match self.sync_ {
Some(scope) => {
let report = self.app_.sync_commands(
client,
ready.application.id,
scope~,
unowned=self.sync_unowned_,
)
for scope in report.scopes {
if !scope.deleted.is_empty() {
let target = match scope.guild_id {
None => "global scope"
Some(guild_id) => "guild \{guild_id}"
}
self.app_.warn(
"command sync deleted \{scope.deleted.join(", ")} in \{target}; use sync_unowned=Keep to preserve undeclared commands",
)
}
}
}
None => ()
}
}
start_services(ctx)
self.collector_.dispatch(event)
self.dispatch_event(spawner, ctx, event)
}
Dispatch(InteractionCreate(interaction) as event) => {
if self.cache_ is Some(cache) {
cache.update(event)
}
match (owner.context, framework) {
(Some(ctx), Some(router)) => {
start_services(ctx)
self.collector_.dispatch(event)
// Processing is another bounded child task. Awaiting it here
// would deadlock handlers waiting for a later component event.
self.spawn_interaction(spawner, router, interaction)
self.dispatch_event(spawner, ctx, event)
}
_ =>
self.app_.warn(
"dropping INTERACTION_CREATE received before shard \{owner.id} READY",
)
}
}
Dispatch(event) => {
let event = match (event, self.cache_) {
(MessageUpdate(update), Some(cache)) =>
@model.Event::MessageUpdate({
..update,
before: cache.message(
update.message.channel_id,
update.message.id,
),
})
_ => event
}
if self.cache_ is Some(cache) {
cache.update(event)
}
self.observe_decode_error(event)
match owner.context {
Some(ctx) => {
start_services(ctx)
self.collector_.dispatch(event)
self.dispatch_event(spawner, ctx, event)
}
None =>
self.app_.warn(
"dropping \{Repr(event_kind(event))} received before shard \{owner.id} READY",
)
}
}
Connected(resumed~) =>
if resumed && owner.context is Some(ctx) {
start_services(ctx)
}
Disconnected(..) => if shutdown_requested.val { break }
FatallyClosed(code~) => {
if shutdown_requested.val {
break
}
close_shards(shard_handles)
raise BotError::FatallyClosed(code~)
}
}
}
})
}