NASP/rpki
84
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add dockerfile

Browse Source
This commit is contained in:
xiuting.xu
2026-04-10 14:29:40 +08:00
parent 23cdad095d
commit c1d3112a45
10 changed files with 257 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
src/bin/rtr_debug_client/main.rs
View File

@@ -1,4 +1,5 @@
use std::env;
use std::future::pending;
use std::io;
use std::path::{Path, PathBuf};
use std::sync::Arc;
@@ -27,6 +28,12 @@ impl<T> AsyncStream for T where T: AsyncRead + AsyncWrite + Unpin + Send {}
type DynStream = Box<dyn AsyncStream>;
type ClientWriter = WriteHalf<DynStream>;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum OutputMode {
Verbose,
SummaryOnly,
}
#[tokio::main]
async fn main() -> io::Result<()> {
let config = Config::from_args()?;
@@ -41,6 +48,7 @@ async fn main() -> io::Result<()> {
config.default_poll_secs
);
println!("keep-after-error: {}", config.keep_after_error);
println!("output : {}", config.output_mode.describe());
match &config.mode {
QueryMode::Reset => {
println!("mode : reset");
@@ -59,10 +67,12 @@ async fn main() -> io::Result<()> {
config.read_timeout_secs,
config.default_poll_secs,
config.keep_after_error,
config.output_mode,
);
let stdin = tokio::io::stdin();
let mut stdin_lines = BufReader::new(stdin).lines();
let mut stdin_closed = false;
loop {
let stream = loop {
@@ -89,7 +99,13 @@ async fn main() -> io::Result<()> {
tokio::pin!(poll_sleep);
tokio::select! {
line = stdin_lines.next_line() => {
line = async {
if stdin_closed {
pending::<io::Result<Option<String>>>().await
} else {
stdin_lines.next_line().await
}
} => {
match line {
Ok(Some(line)) => {
match handle_console_command(
@@ -111,7 +127,8 @@ async fn main() -> io::Result<()> {
}
}
Ok(None) => {
println!("stdin closed, continue network loop.");
stdin_closed = true;
println!("stdin closed, disable console input.");
}
Err(err) => {
eprintln!("read stdin failed: {}", err);
@@ -136,8 +153,11 @@ async fn main() -> io::Result<()> {
) => {
match read_result {
Ok(Ok(pdu)) => {
print_raw_pdu(&pdu.header, &pdu.body);
print_pdu(&pdu.header, &pdu.body);
state.observe_pdu(&pdu.header);
if should_print_pdu(state.output_mode, &pdu.header) {
print_raw_pdu(&pdu.header, &pdu.body);
print_pdu(&pdu.header, &pdu.body);
}
match handle_incoming_pdu(&mut writer, &mut state, &pdu.header, &pdu.body).await {
Ok(()) => {}
Err(err) if should_reconnect(&err) => {
@@ -177,7 +197,13 @@ async fn main() -> io::Result<()> {
loop {
tokio::select! {
_ = &mut reconnect_sleep => break,
line = stdin_lines.next_line() => {
line = async {
if stdin_closed {
pending::<io::Result<Option<String>>>().await
} else {
stdin_lines.next_line().await
}
} => {
match line {
Ok(Some(line)) => {
match handle_console_command(&line, None, &mut state).await {
@@ -197,7 +223,8 @@ async fn main() -> io::Result<()> {
}
}
Ok(None) => {
println!("stdin closed, continue reconnect loop.");
stdin_closed = true;
println!("stdin closed, disable console input.");
}
Err(err) => {
eprintln!("read stdin failed: {}", err);
@@ -326,6 +353,16 @@ async fn handle_incoming_pdu(
);
}
if state.output_mode == OutputMode::SummaryOnly
&& state.skipped_payload_pdu_count_in_round > 0
{
println!(
"summary : skipped {} payload PDUs in this response",
state.skipped_payload_pdu_count_in_round
);
state.skipped_payload_pdu_count_in_round = 0;
}
println!("received EndOfData, keep connection open.");
println!();
}
@@ -370,6 +407,7 @@ async fn handle_incoming_pdu(
state.current_session_id = None;
state.serial = None;
state.last_error_code = None;
state.skipped_payload_pdu_count_in_round = 0;
send_reset_query(writer, state.version).await?;
state.schedule_next_poll();
println!();
@@ -604,6 +642,21 @@ async fn handle_console_command(
return Ok(true);
}
["output"] => {
println!("current output mode: {}", state.output_mode.describe());
println!("skipped payload PDUs: {}", state.skipped_payload_pdu_count);
}
["output", "verbose"] => {
state.output_mode = OutputMode::Verbose;
println!("updated output mode to verbose");
}
["output", "summary"] => {
state.output_mode = OutputMode::SummaryOnly;
println!("updated output mode to summary");
}
_ => {
println!("unknown command: {}", line);
print_help();
@@ -629,6 +682,9 @@ fn print_help() {
println!(" poll pause pause auto polling");
println!(" poll resume resume auto polling");
println!(" keep-after-error show current keep-after-error setting");
println!(" output show current output mode");
println!(" output verbose print all PDUs");
println!(" output summary suppress payload PDU details");
println!(" quit exit client");
println!();
}
@@ -648,6 +704,8 @@ fn print_state(state: &ClientState) {
println!(" poll_source : {}", state.poll_interval_source());
println!(" last_error_code : {:?}", state.last_error_code);
println!(" keep_after_error : {}", state.keep_after_error);
println!(" output_mode : {}", state.output_mode.describe());
println!(" skipped_payloads : {}", state.skipped_payload_pdu_count);
println!(" poll_paused : {}", state.poll_paused);
println!();
}
@@ -664,6 +722,9 @@ struct ClientState {
expire: Option<u32>,
last_error_code: Option<u16>,
keep_after_error: bool,
output_mode: OutputMode,
skipped_payload_pdu_count: u64,
skipped_payload_pdu_count_in_round: u64,
read_timeout_secs: u64,
default_poll_secs: u64,
@@ -679,6 +740,7 @@ impl ClientState {
read_timeout_secs: u64,
default_poll_secs: u64,
keep_after_error: bool,
output_mode: OutputMode,
) -> Self {
Self {
version,
@@ -690,6 +752,9 @@ impl ClientState {
expire: None,
last_error_code: None,
keep_after_error,
output_mode,
skipped_payload_pdu_count: 0,
skipped_payload_pdu_count_in_round: 0,
read_timeout_secs,
default_poll_secs,
next_poll_deadline: Instant::now() + Duration::from_secs(default_poll_secs),
@@ -770,6 +835,14 @@ impl ClientState {
false
}
}
fn observe_pdu(&mut self, header: &PduHeader) {
if self.output_mode == OutputMode::SummaryOnly && is_payload_pdu(header) {
self.skipped_payload_pdu_count = self.skipped_payload_pdu_count.saturating_add(1);
self.skipped_payload_pdu_count_in_round =
self.skipped_payload_pdu_count_in_round.saturating_add(1);
}
}
}
#[derive(Debug)]
@@ -781,6 +854,7 @@ struct Config {
default_poll_secs: u64,
transport: TransportConfig,
keep_after_error: bool,
output_mode: OutputMode,
}
impl Config {
@@ -791,6 +865,7 @@ impl Config {
let mut read_timeout_secs = DEFAULT_READ_TIMEOUT_SECS;
let mut default_poll_secs = DEFAULT_POLL_INTERVAL_SECS;
let mut keep_after_error = false;
let mut output_mode = OutputMode::Verbose;
while let Some(arg) = args.next() {
match arg.as_str() {
@@ -841,6 +916,9 @@ impl Config {
"--keep-after-error" => {
keep_after_error = true;
}
"--summary-only" => {
output_mode = OutputMode::SummaryOnly;
}
_ if arg.starts_with("--") => {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
@@ -924,10 +1002,34 @@ impl Config {
default_poll_secs,
transport,
keep_after_error,
output_mode,
})
}
}
impl OutputMode {
fn describe(self) -> &'static str {
match self {
Self::Verbose => "verbose",
Self::SummaryOnly => "summary",
}
}
}
fn is_payload_pdu(header: &PduHeader) -> bool {
matches!(
header.pdu_type(),
PduType::Ipv4Prefix | PduType::Ipv6Prefix | PduType::RouterKey | PduType::Aspa
)
}
fn should_print_pdu(output_mode: OutputMode, header: &PduHeader) -> bool {
match output_mode {
OutputMode::Verbose => true,
OutputMode::SummaryOnly => !is_payload_pdu(header),
}
}
#[derive(Debug, Clone)]
enum TransportConfig {
Tcp,