TLA  Line data    Source code

       1                 : #![deny(clippy::undocumented_unsafe_blocks)]
       2                 : 
       3                 : use std::convert::Infallible;
       4                 : 
       5                 : use anyhow::{bail, Context};
       6                 : use tokio::task::JoinError;
       7                 : use tokio_util::sync::CancellationToken;
       8                 : use tracing::warn;
       9                 : 
      10                 : pub mod auth;
      11                 : pub mod cache;
      12                 : pub mod cancellation;
      13                 : pub mod compute;
      14                 : pub mod config;
      15                 : pub mod console;
      16                 : pub mod context;
      17                 : pub mod error;
      18                 : pub mod http;
      19                 : pub mod logging;
      20                 : pub mod metrics;
      21                 : pub mod parse;
      22                 : pub mod protocol2;
      23                 : pub mod proxy;
      24                 : pub mod rate_limiter;
      25                 : pub mod sasl;
      26                 : pub mod scram;
      27                 : pub mod serverless;
      28                 : pub mod stream;
      29                 : pub mod url;
      30                 : pub mod usage_metrics;
      31                 : pub mod waiters;
      32                 : 
      33                 : /// Handle unix signals appropriately.
      34 CBC          23 : pub async fn handle_signals(token: CancellationToken) -> anyhow::Result<Infallible> {
      35                 :     use tokio::signal::unix::{signal, SignalKind};
      36                 : 
      37              23 :     let mut hangup = signal(SignalKind::hangup())?;
      38              23 :     let mut interrupt = signal(SignalKind::interrupt())?;
      39              23 :     let mut terminate = signal(SignalKind::terminate())?;
      40                 : 
      41              46 :     loop {
      42              69 :         tokio::select! {
      43              69 :             // Hangup is commonly used for config reload.
      44              69 :             _ = hangup.recv() => {
      45              69 :                 warn!("received SIGHUP; config reload is not supported");
      46              69 :             }
      47              69 :             // Shut down the whole application.
      48              69 :             _ = interrupt.recv() => {
      49              69 :                 warn!("received SIGINT, exiting immediately");
      50              69 :                 bail!("interrupted");
      51              69 :             }
      52              69 :             _ = terminate.recv() => {
      53              69 :                 warn!("received SIGTERM, shutting down once all existing connections have closed");
      54              69 :                 token.cancel();
      55              69 :             }
      56              69 :         }
      57              46 :     }
      58 UBC           0 : }
      59                 : 
      60                 : /// Flattens `Result<Result<T>>` into `Result<T>`.
      61 CBC          67 : pub fn flatten_err<T>(r: Result<anyhow::Result<T>, JoinError>) -> anyhow::Result<T> {
      62              67 :     r.context("join error").and_then(|x| x)
      63              67 : }