rippinrobr

let sys = actix::System::new("conspiracies-api");
// Start 3 parallel db executors
let addr = SyncArbiter::start(3, || {
    DbExecutor(SqliteConnection::establish("database/conspiracies.sqlite3").unwrap())
});

..... Route definitions here .....

let _ = sys.run();

rippinrobr

// Send message to `DbExecutor` actor
req.state().db.send(GetConspiracy{page_id: page_id.to_owned()})
    .from_err()
    .and_then(|res| {
        match res {
            Ok(conspiracy) => Ok(HttpResponse::Ok().json(conspiracy)),
            Err(_) => Ok(HttpResponse::NotFound().into())
        }
    })
.responder()

rippinrobr

/// The DbExecutor is contains the database connection which will be used by all of the
/// message handlers
pub struct DbExecutor(pub SqliteConnection);

impl Actor for DbExecutor {
   type Context = SyncContext<Self>;
}

/// Sent to the DbExecutor to fetch the conspiracy with the given page_id
pub struct GetConspiracy {

rippinrobr

App::with_state(State{db: addr.clone()})
  // the default_resource call here uses a closure
  // to return 405 Method Not Allowed error for 
  // all routes
  .default_resource(|r| {
     r.route().filter(pred::Not(pred::Get()))
         .f(|req| HttpResponse::MethodNotAllowed());
  })
  .resource("/", |r| r.method(http::Method::GET).f(index))
  .resource("/conspiracies/{page_id}", |r| r.method(http::Method::GET).a(get_conspiracies_by_id))

rippinrobr

/// returns the conspiracy by the given id
fn get_conspiracies_by_id(req: HttpRequest<State>) -> impl Future<Item=HttpResponse, Error=Error> {
    let page_id = &req.match_info()["page_id"];

    // Send message to `DbExecutor` actor
    req.state().db.send(GetConspiracy{page_id: page_id.to_owned()})
        .from_err()
        .and_then(|res| {
            match res {
                Ok(conspiracy) => Ok(HttpResponse::Ok().json(conspiracy)),

rippinrobr

let links =  db::get_links_to_process(&conn, batch_size);
WikiRepo::get_conspiracies(&c, links, title.to_string(), |p2, categories| {
    match db::add_conspiracy(&conn, &p2) {
        Err(e) => println!("SAVE ERROR: {} {} {}", e ,p2.title, p2.page_id),
        Ok(_) => {
            let title = &p2.title;
            db::add_categories(&conn, categories).unwrap();
            db::mark_link_as_processed(&conn, title).expect(&format!("A problem occurred when marking the link '{}' as processed",title));
            println!("Added: {} {}", title, p2.page_id)
        }

rippinrobr

let page = client.page_from_title(title.to_string()); 
let links_iter = page.get_links().expect("unable to get the links");
for (i, l) in links_iter.enumerate() {
    let lp = LinkProcessed {
        title: l.title,
        processed: 0,
    };
    save_action(lp);
}

rippinrobr

pub fn get_page_links<'a, F>(client: &'a wikipedia::Wikipedia::<wikipedia::http::default::Client>, 
                             title: String, 
                             save_action: F)
        where F: Fn(LinkProcessed) {

rippinrobr

if _matches.is_present("get-links") {
    WikiRepo::get_page_links(&c, title.to_string(), |link| {
        match db::add_link_process(&conn, &link) {
            Err(e) => println!("SAVE ERROR: {} {}", e ,link.title),
            Ok(_) => println!("Added: {}", link.title)
        };
    });
}

rippinrobr

pub fn mark_link_as_processed(conn: &SqliteConnection, link_title: &str) ->Result<usize, diesel::result::Error> {
    let u_stmt = format!("UPDATE links_processed SET processed=1 WHERE title='{}';", link_title.replace("'", "''"));
    let update = sql::<Bool>(&u_stmt);
    update.execute(conn)
}