mirror of
https://github.com/neondatabase/neon.git
synced 2026-05-27 01:50:38 +00:00
99399c112a
Walreceiver is a per-timeline abstraction. Move it there to reflect
the hierarchy of abstractions and task_mgr tasks.
The code that sets up the global storage_broker client
is not timeline-scoped. So, break it out into a separate module.
The motivation for this change is to prepare the code base for replacing
the task_mgr global task registry with a more ownership-oriented
approach to manage task lifetimes.
I removed TaskStateUpdate::Init because, after doing the changes,
rustc warned that it was never constructed.
A quick search through the commit history shows that this
has always been true since
commit fb68d01449
Author: Dmitry Rodionov <dmitry@neon.tech>
Date: Mon Sep 26 23:57:02 2022 +0300
Preserve task result in TaskHandle by keeping join handle around (#2521)
So, the warning is not an indication of some accidental code removal.
This is PR: https://github.com/neondatabase/neon/pull/3456
49 lines
1.4 KiB
Rust
49 lines
1.4 KiB
Rust
//! The broker client instance of the pageserver, created during pageserver startup.
|
|
//! Used by each timelines' [`walreceiver`].
|
|
|
|
use crate::config::PageServerConf;
|
|
|
|
use anyhow::Context;
|
|
use once_cell::sync::OnceCell;
|
|
use storage_broker::BrokerClientChannel;
|
|
use tracing::*;
|
|
|
|
static BROKER_CLIENT: OnceCell<BrokerClientChannel> = OnceCell::new();
|
|
|
|
///
|
|
/// Initialize the broker client. This must be called once at page server startup.
|
|
///
|
|
pub async fn init_broker_client(conf: &'static PageServerConf) -> anyhow::Result<()> {
|
|
let broker_endpoint = conf.broker_endpoint.clone();
|
|
|
|
// Note: we do not attempt connecting here (but validate endpoints sanity).
|
|
let broker_client =
|
|
storage_broker::connect(broker_endpoint.clone(), conf.broker_keepalive_interval).context(
|
|
format!(
|
|
"Failed to create broker client to {}",
|
|
&conf.broker_endpoint
|
|
),
|
|
)?;
|
|
|
|
if BROKER_CLIENT.set(broker_client).is_err() {
|
|
panic!("broker already initialized");
|
|
}
|
|
|
|
info!(
|
|
"Initialized broker client with endpoints: {}",
|
|
broker_endpoint
|
|
);
|
|
Ok(())
|
|
}
|
|
|
|
///
|
|
/// Get a handle to the broker client
|
|
///
|
|
pub fn get_broker_client() -> &'static BrokerClientChannel {
|
|
BROKER_CLIENT.get().expect("broker client not initialized")
|
|
}
|
|
|
|
pub fn is_broker_client_initialized() -> bool {
|
|
BROKER_CLIENT.get().is_some()
|
|
}
|