jayhuang75

HdfsAdmin admin = new HdfsAdmin(URI.create(args[0]), new Configuration());

DFSInotifyEventInputStream eventStream = admin.getInotifyEventStream(lastReadTxid);

while (true) {
  EventBatch batch = eventStream.take();
  System.out.println("TxId = " + batch.getTxid());
  
  for (Event event : batch.getEvents()) {
    System.out.println("event type = " + event.getEventType());

jayhuang75

pub struct DbContext<'a> {
    pub conn: &'a SqliteConnection,
}

impl<'a> DbContext<'a> {
    pub fn new(conn: &'a SqliteConnection) -> Self {
        return DbContext { conn };
    }
    pub fn count(&mut self) -> Result<i64, Error> {
        let app_count: i64 = apps.count().get_result(self.conn)?;

jayhuang75

#[tokio::main]
async fn main() {
    // prepare local env
    dotenv().ok();

    // set the interval for every 20s
    let mut interval = time::interval(time::Duration::from_secs(20));

    loop {
        // wait every 20s

jayhuang75

.ONESHELL:
RELEASE_BIN=rust-hadoop-batch-control
RELEASE_BIN_PATH_LINUX=target/release
RELEASE_BIN_PATH_DARWIN=target/x86_64-unknown-linux-musl/release
HADOOP_NAMENODE=namenode
DB_LOCATION=db
DB_NAME=apps.db

.PHONY: darwin
darwin: 

jayhuang75

root@c1f89f553e01:/# DATABASE_URL=db/apps.db ./rust-hadoop-batch-control 
[main.rs] total apps in db : 13
[main.rs] current log total count: 13
[main.rs] time elapsed in run() is: 1.879390475s
[main.rs] total apps in db : 13
[main.rs] current log total count: 13
[main.rs] time elapsed in run() is: 1.860675949s
[main.rs] total apps in db : 13
[main.rs] current log total count: 13
[main.rs] time elapsed in run() is: 1.818562553s

jayhuang75

pub async fn fetch(url: &str, dir_name: &str, file_name: &str) -> Result<String, Box<dyn Error>> {
    // application start time
    let start = Instant::now();

    // create folder if not exist
    fs::create_dir_all(dir_name)?;

    // download the file
    let resp = reqwest::get(url).await?;

jayhuang75

pub async fn run(file: &str) -> Result<Vec<alert::CovidStats>, Box<dyn Error>> {
    let mut reader = csv::ReaderBuilder::new()
        .has_headers(true)
        .from_path(file)?;

    let today_covid_report: Vec<alert::CovidStats> = reader
        .records()
        .into_iter()
        .par_bridge()
        .map(|line| {

jayhuang75

#[async_trait(?Send)]
pub trait Producer {
    async fn run(&self, covid: &CovidReport) -> Result<(), Box<dyn Error>>;
}

pub struct CovidReport {
    pub report: Vec<CovidStats>,
    // any type inside a Box must implement the Producer trait.
    producer: Box<dyn Producer>,
}

jayhuang75

impl Producer for Sms {
    async fn run(&self, covid: &CovidReport) -> Result<(), Box<dyn Error>> {
        // in the feature will loop the phone number in some of the use case
        let to = env::var("TWILIO_TO_NUM").unwrap().as_str().to_string();

        let client = Client::new(&*TWILIO_ACCOUNT_ID, &*TWILIO_AUTH_TOKEN);

        let body = &format!(
            "Today Ontario has {} new cases, number of death {}. ",
            covid.report[0].numtoday, covid.report[0].numdeathstoday

jayhuang75

    // application start time
    let start = Instant::now();
    let now: Date<Local> = Local::now().date();
    info!("[job] start at {:?}", now);

    info!("[job] download url : {:?}", &*DOWNLOAD_FILE_URL);

    // download the file
    let download_file_name = format!("{}-{}", now, *DOWNLOAD_FILE_NAME);
    let file_path: String = download::fetch(