Merge pull request #1 from AdExNetwork/improvements-analytics-recorder

Improvements on the code

Author: samparsky

sentry/src/analytics_recorder.rs

@@ -10,15 +10,13 @@ use slog::{error, Logger};
 pub fn get_payout(channel: &Channel, event: &Event) -> BigNum {
     match event {
         Event::Impression { .. } => channel.spec.min_per_impression.clone(),
-        Event::Click { .. } => {
-            if let Some(pricing) = channel.spec.pricing_bounds.clone() {
-                if let Some(click) = pricing.click {
-                    return click.min;
-                }
-            }
-
-            Default::default()
-        }
+        Event::Click { .. } => channel
+            .spec
+            .pricing_bounds
+            .as_ref()
+            .and_then(|pricing_bound| pricing_bound.click.as_ref())
+            .map(|click| click.min.clone())
+            .unwrap_or_default(),
         _ => Default::default(),
     }
 }
@@ -84,7 +82,7 @@ pub async fn record(
                     .ignore();
                 }
 
-                if let Some(country) = session.country.clone() {
+                if let Some(country) = &session.country {
                     db.zincr(
                         format!(
                             "{}:{}:{}:{}",
@@ -129,6 +127,6 @@ pub async fn record(
         });
 
     if let Err(err) = db.query_async::<_, Option<String>>(&mut conn).await {
-        error!(&logger, "Server error: {}", err; "module" => "analytics-recorder");
+        error!(&logger, "Redis Database error: {}", err; "module" => "analytics-recorder");
     }
 }

sentry/src/db/event_aggregate.rs

@@ -55,7 +55,7 @@ pub async fn latest_new_state(
 pub async fn latest_heartbeats(
     pool: &DbPool,
     channel_id: &ChannelId,
-    validator_id: ValidatorId,
+    validator_id: &ValidatorId,
 ) -> Result<Vec<HeartbeatValidatorMessage>, RunError<bb8_postgres::tokio_postgres::Error>> {
     pool
     .run(move |connection| {

sentry/src/event_aggregator.rs

@@ -140,13 +140,12 @@ impl EventAggregator {
             .for_each(|ev| event_reducer::reduce(&record.channel, &mut record.aggregate, ev));
 
         if ANALYTICS_RECORDER.is_some() {
-            let logger = app.logger.clone();
             tokio::spawn(analytics_recorder::record(
                 redis.clone(),
                 record.channel.clone(),
                 session.clone(),
                 events.to_owned().to_vec(),
-                logger,
+                app.logger.clone(),
             ));
         }
 
@@ -155,13 +154,7 @@ impl EventAggregator {
         drop(channel_recorder);
 
         if aggr_throttle == 0 {
-            store(
-                &app.pool,
-                &channel_id,
-                &app.logger.clone(),
-                recorder.clone(),
-            )
-            .await;
+            store(&app.pool, &channel_id, &app.logger, recorder.clone()).await;
         }
 
         Ok(())

sentry/src/routes/channel.rs

@@ -138,8 +138,8 @@ pub async fn last_approved<A: Adapter>(
     let heartbeats = if query.with_heartbeat.is_some() {
         let result = try_join_all(
             validators
-                .into_iter()
-                .map(|validator| latest_heartbeats(&app.pool, &channel_id, validator.id.clone())),
+                .iter()
+                .map(|validator| latest_heartbeats(&app.pool, &channel_id, &validator.id)),
         )
         .await?;
         Some(result.into_iter().flatten().collect::<Vec<_>>())