refactor: migrate to clap from structopt

README.md

@@ -51,7 +51,7 @@ Here is a basic [stream consumer](https://github.com/SeaQL/sea-streamer/tree/mai
 async fn main() -> Result<()> {
     env_logger::init();
 
-    let Args { stream } = Args::from_args();
+    let Args { stream } = Args::parse();
 
     let streamer = SeaStreamer::connect(stream.streamer(), Default::default()).await?;
 
@@ -76,7 +76,7 @@ Here is a basic [stream producer](https://github.com/SeaQL/sea-streamer/tree/mai
 async fn main() -> Result<()> {
     env_logger::init();
 
-    let Args { stream } = Args::from_args();
+    let Args { stream } = Args::parse();
 
     let streamer = SeaStreamer::connect(stream.streamer(), Default::default()).await?;
 
@@ -105,7 +105,7 @@ See also other [advanced stream processors](https://github.com/SeaQL/sea-streame
 async fn main() -> Result<()> {
     env_logger::init();
 
-    let Args { input, output } = Args::from_args();
+    let Args { input, output } = Args::parse();
 
     let streamer = SeaStreamer::connect(input.streamer(), Default::default()).await?;
     let options = SeaConsumerOptions::new(ConsumerMode::RealTime);

benchmark/src/bin/baseline.rs

@@ -1,11 +1,11 @@
 use anyhow::Result;
 use sea_streamer::{runtime::sleep, StreamUrl};
 use std::time::Duration;
-use structopt::StructOpt;
+use clap::Parser;
 
-#[derive(Debug, StructOpt)]
+#[derive(Debug, Parser)]
 struct Args {
-    #[structopt(
+    #[clap(
         long,
         help = "Streamer URI with stream key, i.e. try `kafka://localhost:9092/my_topic`",
         env = "STREAM_URL"
@@ -18,7 +18,7 @@ struct Args {
 async fn main() -> Result<()> {
     env_logger::init();
 
-    let Args { stream } = Args::from_args();
+    let Args { stream } = Args::parse();
     std::hint::black_box(stream);
 
     for i in 0..100_000 {

benchmark/src/bin/consumer.rs

@@ -3,11 +3,11 @@ use sea_streamer::{
     Buffer, Consumer, ConsumerMode, ConsumerOptions, Message, SeaConsumer, SeaConsumerOptions,
     SeaMessage, SeaStreamReset, SeaStreamer, StreamUrl, Streamer,
 };
-use structopt::StructOpt;
+use clap::Parser;
 
-#[derive(Debug, StructOpt)]
+#[derive(Debug, Parser)]
 struct Args {
-    #[structopt(
+    #[clap(
         long,
         help = "Streamer URI with stream key(s), i.e. try `kafka://localhost:9092/my_topic`",
         env = "STREAM_URL"
@@ -20,7 +20,7 @@ struct Args {
 async fn main() -> Result<()> {
     env_logger::init();
 
-    let Args { stream } = Args::from_args();
+    let Args { stream } = Args::parse();
 
     let streamer = SeaStreamer::connect(stream.streamer(), Default::default()).await?;
 

benchmark/src/bin/producer.rs

@@ -1,11 +1,11 @@
 use anyhow::Result;
 use sea_streamer::{Producer, SeaProducer, SeaStreamer, StreamUrl, Streamer};
 use std::time::Duration;
-use structopt::StructOpt;
+use clap::Parser;
 
-#[derive(Debug, StructOpt)]
+#[derive(Debug, Parser)]
 struct Args {
-    #[structopt(
+    #[clap(
         long,
         help = "Streamer URI with stream key, i.e. try `kafka://localhost:9092/my_topic`",
         env = "STREAM_URL"
@@ -18,7 +18,7 @@ struct Args {
 async fn main() -> Result<()> {
     env_logger::init();
 
-    let Args { stream } = Args::from_args();
+    let Args { stream } = Args::parse();
 
     let streamer = SeaStreamer::connect(stream.streamer(), Default::default()).await?;
 

benchmark/src/bin/relay.rs

@@ -1,21 +1,21 @@
 use anyhow::Result;
 use sea_streamer::stdio::StdioStreamer;
 use sea_streamer::{Consumer, Message, Producer, StreamKey, Streamer, StreamerUri};
-use structopt::StructOpt;
+use clap::Parser;
 
-#[derive(Debug, StructOpt)]
+#[derive(Debug, Parser)]
 struct Args {
-    #[structopt(long, help = "Stream key of input")]
+    #[clap(long, help = "Stream key of input")]
     input: StreamKey,
-    #[structopt(long, help = "Stream key of output")]
+    #[clap(long, help = "Stream key of output")]
     output: StreamKey,
 }
 
 #[tokio::main]
 async fn main() -> Result<()> {
     env_logger::init();
 
-    let Args { input, output } = Args::from_args();
+    let Args { input, output } = Args::parse();
 
     let streamer = StdioStreamer::connect(StreamerUri::zero(), Default::default()).await?;
     let consumer = streamer

examples/src/bin/blocking.rs

@@ -1,22 +1,22 @@
 use anyhow::Result;
 use flume::bounded;
 use std::time::Duration;
-use structopt::StructOpt;
+use clap::Parser;
 
 use sea_streamer::{
     runtime::{sleep, spawn_task},
     Buffer, Consumer, ConsumerMode, ConsumerOptions, Message, Producer, SeaConsumer,
     SeaConsumerOptions, SeaMessage, SeaProducer, SeaStreamer, SharedMessage, StreamUrl, Streamer,
 };
 
-#[derive(Debug, StructOpt)]
+#[derive(Debug, Parser)]
 struct Args {
-    #[structopt(
+    #[clap(
         long,
         help = "Streamer URI with stream key(s), i.e. try `kafka://localhost:9092/my_topic`"
     )]
     input: StreamUrl,
-    #[structopt(
+    #[clap(
         long,
         help = "Streamer URI with stream key, i.e. try `stdio:///my_stream`"
     )]
@@ -30,7 +30,7 @@ const NUM_THREADS: usize = 4; // Every one has at least 2 cores with 2 hyperthre
 async fn main() -> Result<()> {
     env_logger::init();
 
-    let Args { input, output } = Args::from_args();
+    let Args { input, output } = Args::parse();
 
     // The queue
     let (sender, receiver) = bounded(1024);

examples/src/bin/buffered.rs

@@ -1,22 +1,22 @@
 use anyhow::Result;
 use flume::bounded;
 use std::time::Duration;
-use structopt::StructOpt;
+use clap::Parser;
 
 use sea_streamer::{
     runtime::{sleep, spawn_task},
     Buffer, Consumer, ConsumerMode, ConsumerOptions, Message, Producer, SeaConsumer,
     SeaConsumerOptions, SeaMessage, SeaProducer, SeaStreamer, SharedMessage, StreamUrl, Streamer,
 };
 
-#[derive(Debug, StructOpt)]
+#[derive(Debug, Parser)]
 struct Args {
-    #[structopt(
+    #[clap(
         long,
         help = "Streamer URI with stream key(s), i.e. try `kafka://localhost:9092/my_topic`"
     )]
     input: StreamUrl,
-    #[structopt(
+    #[clap(
         long,
         help = "Streamer URI with stream key, i.e. try `stdio:///my_stream`"
     )]
@@ -28,7 +28,7 @@ struct Args {
 async fn main() -> Result<()> {
     env_logger::init();
 
-    let Args { input, output } = Args::from_args();
+    let Args { input, output } = Args::parse();
 
     // The queue
     let (sender, receiver) = bounded(1024);

examples/src/bin/consumer.rs

@@ -3,11 +3,11 @@ use sea_streamer::{
     Buffer, Consumer, ConsumerMode, ConsumerOptions, Message, SeaConsumer, SeaConsumerOptions,
     SeaMessage, SeaStreamReset, SeaStreamer, StreamUrl, Streamer,
 };
-use structopt::StructOpt;
+use clap::Parser;
 
-#[derive(Debug, StructOpt)]
+#[derive(Debug, Parser)]
 struct Args {
-    #[structopt(
+    #[clap(
         long,
         help = "Streamer URI with stream key(s), i.e. try `kafka://localhost:9092/my_topic`",
         env = "STREAM_URL"
@@ -20,7 +20,7 @@ struct Args {
 async fn main() -> Result<()> {
     env_logger::init();
 
-    let Args { stream } = Args::from_args();
+    let Args { stream } = Args::parse();
 
     let streamer = SeaStreamer::connect(stream.streamer(), Default::default()).await?;
 

examples/src/bin/processor.rs

@@ -3,16 +3,16 @@ use sea_streamer::{
     Buffer, Consumer, ConsumerMode, ConsumerOptions, Message, Producer, SeaConsumer,
     SeaConsumerOptions, SeaMessage, SeaProducer, SeaStreamer, StreamUrl, Streamer,
 };
-use structopt::StructOpt;
+use clap::Parser;
 
-#[derive(Debug, StructOpt)]
+#[derive(Debug, Parser)]
 struct Args {
-    #[structopt(
+    #[clap(
         long,
         help = "Streamer URI with stream key(s), i.e. try `kafka://localhost:9092/my_topic`"
     )]
     input: StreamUrl,
-    #[structopt(
+    #[clap(
         long,
         help = "Streamer URI with stream key, i.e. try `stdio:///my_stream`"
     )]
@@ -24,7 +24,7 @@ struct Args {
 async fn main() -> Result<()> {
     env_logger::init();
 
-    let Args { input, output } = Args::from_args();
+    let Args { input, output } = Args::parse();
 
     let streamer = SeaStreamer::connect(input.streamer(), Default::default()).await?;
     let options = SeaConsumerOptions::new(ConsumerMode::RealTime);

examples/src/bin/producer.rs

@@ -1,11 +1,11 @@
 use anyhow::Result;
 use sea_streamer::{Producer, SeaProducer, SeaStreamer, StreamUrl, Streamer};
 use std::time::Duration;
-use structopt::StructOpt;
+use clap::Parser;
 
-#[derive(Debug, StructOpt)]
+#[derive(Debug, Parser)]
 struct Args {
-    #[structopt(
+    #[clap(
         long,
         help = "Streamer URI with stream key, i.e. try `kafka://localhost:9092/my_topic`",
         env = "STREAM_URL"
@@ -18,7 +18,7 @@ struct Args {
 async fn main() -> Result<()> {
     env_logger::init();
 
-    let Args { stream } = Args::from_args();
+    let Args { stream } = Args::parse();
 
     let streamer = SeaStreamer::connect(stream.streamer(), Default::default()).await?;
 

examples/src/bin/resumable.rs

@@ -7,18 +7,18 @@ use sea_streamer::{
     Streamer,
 };
 use std::time::Duration;
-use structopt::StructOpt;
+use clap::Parser;
 
 const TRANSACTION: bool = true;
 
-#[derive(Debug, StructOpt)]
+#[derive(Debug, Parser)]
 struct Args {
-    #[structopt(
+    #[clap(
         long,
         help = "Streamer URI with stream key(s), i.e. try `kafka://localhost:9092/my_topic`"
     )]
     input: StreamUrl,
-    #[structopt(
+    #[clap(
         long,
         help = "Streamer URI with stream key, i.e. try `stdio:///my_stream`"
     )]
@@ -30,7 +30,7 @@ struct Args {
 async fn main() -> Result<()> {
     env_logger::init();
 
-    let Args { input, output } = Args::from_args();
+    let Args { input, output } = Args::parse();
 
     let streamer = SeaStreamer::connect(input.streamer(), Default::default()).await?;
     let mut options = SeaConsumerOptions::new(ConsumerMode::Resumable);

sea-streamer-file/src/bin/clock.rs

@@ -2,13 +2,13 @@ use anyhow::{anyhow, Result};
 use sea_streamer_file::{FileId, FileStreamer};
 use sea_streamer_types::{Producer, StreamKey, Streamer};
 use std::time::Duration;
-use structopt::StructOpt;
+use clap::Parser;
 
-#[derive(Debug, StructOpt)]
+#[derive(Debug, Parser)]
 struct Args {
-    #[structopt(long, help = "Stream to this file")]
+    #[clap(long, help = "Stream to this file")]
     file: FileId,
-    #[structopt(long, parse(try_from_str = parse_duration), help = "Period of the clock. e.g. 1s, 100ms")]
+    #[clap(long, parse(try_from_str = parse_duration), help = "Period of the clock. e.g. 1s, 100ms")]
     interval: Duration,
 }
 
@@ -28,7 +28,7 @@ fn parse_duration(src: &str) -> Result<Duration> {
 async fn main() -> Result<()> {
     env_logger::init();
 
-    let Args { file, interval } = Args::from_args();
+    let Args { file, interval } = Args::parse();
 
     let stream_key = StreamKey::new("clock")?;
     let streamer = FileStreamer::connect(file.to_streamer_uri()?, Default::default()).await?;

sea-streamer-file/src/bin/decoder.rs

@@ -21,15 +21,15 @@ use sea_streamer_file::{
 };
 use sea_streamer_types::{Buffer, Message, TIMESTAMP_FORMAT};
 use std::str::FromStr;
-use structopt::StructOpt;
+use clap::Parser;
 
-#[derive(StructOpt)]
+#[derive(Parser)]
 struct Args {
-    #[structopt(long, help = "Decode this file")]
+    #[clap(long, help = "Decode this file")]
     file: FileId,
-    #[structopt(long, help = "If set, skip printing the payload")]
+    #[clap(long, help = "If set, skip printing the payload")]
     header_only: bool,
-    #[structopt(long, help = "The output format", default_value = "log")]
+    #[clap(long, help = "The output format", default_value = "log")]
     format: Format,
 }
 
@@ -46,7 +46,7 @@ async fn main() -> Result<()> {
         file,
         header_only,
         format,
-    } = Args::from_args();
+    } = Args::parse();
 
     let mut source = MessageSource::new(file, StreamMode::Replay).await?;
 

sea-streamer-file/src/bin/sink.rs

@@ -2,13 +2,13 @@ use anyhow::{anyhow, Result};
 use sea_streamer_file::{FileId, MessageSink, DEFAULT_BEACON_INTERVAL, DEFAULT_FILE_SIZE_LIMIT};
 use sea_streamer_types::{MessageHeader, OwnedMessage, ShardId, StreamKey, Timestamp};
 use std::time::Duration;
-use structopt::StructOpt;
+use clap::Parser;
 
-#[derive(Debug, StructOpt)]
+#[derive(Debug, Parser)]
 struct Args {
-    #[structopt(long, help = "Stream to this file")]
+    #[clap(long, help = "Stream to this file")]
     file: FileId,
-    #[structopt(long, parse(try_from_str = parse_duration), help = "Period of the clock. e.g. 1s, 100ms")]
+    #[clap(long, parse(try_from_str = parse_duration), help = "Period of the clock. e.g. 1s, 100ms")]
     interval: Duration,
 }
 
@@ -28,7 +28,7 @@ fn parse_duration(src: &str) -> Result<Duration> {
 async fn main() -> Result<()> {
     env_logger::init();
 
-    let Args { file, interval } = Args::from_args();
+    let Args { file, interval } = Args::parse();
     let mut sink = MessageSink::new(
         file.clone(),
         DEFAULT_BEACON_INTERVAL,