Decorate MainPings and CrashPings with a parse() method

src/main/scala/com/mozilla/telemetry/streaming/ErrorAggregator.scala

@@ -141,7 +141,8 @@ object ErrorAggregator {
     dimensions.build
   }
 
-  private def parseCrashPing(ping: CrashPing): Tuple1[Row] = {
+  class ParsableCrashPing(ping: CrashPing) {
+    def parse(): Tuple1[Row] = {
       // Non-main crashes are already retrieved from main pings
       if(!ping.isMain()) return Tuple1(null)
 
@@ -151,8 +152,11 @@ object ErrorAggregator {
       stats("main_crashes") = Some(1)
       Tuple1(RowBuilder.merge(dimensions, stats.build))
     }
+  }
+  implicit def ParsableCrashPingToCrashPing(ping: CrashPing) = new ParsableCrashPing(ping)
 
-  private def parseMainPing(ping: MainPing): Tuple1[Row] = {
+  class ParsableMainPing(ping: MainPing) {
+    def parse(): Tuple1[Row] = {
       // If a main ping has no usage hours discard it.
       val usageHours = ping.usageHours()
       if (usageHours == 0) return Tuple1(null)
@@ -172,6 +176,9 @@ object ErrorAggregator {
 
       Tuple1(RowBuilder.merge(dimensions, stats.build))
     }
+  }
+  implicit def ParsableCrashPingToMainPing(ping: MainPing) = new ParsableMainPing(ping)
+
   /*
    We can't use an Option[Row] because entire rows cannot be null in Spark SQL. The best we can do is to resort to Tuple1[Row].
    See https://github.com/apache/spark/blob/38b9e69623c14a675b14639e8291f5d29d2a0bc3/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala#L53
@@ -185,9 +192,9 @@ object ErrorAggregator {
       return Tuple1(null)
     }
     if(docType == "crash") {
-      parseCrashPing(messageToCrashPing(message))
+      messageToCrashPing(message).parse()
     } else {
-      parseMainPing(messageToMainPing(message))
+      messageToMainPing(message).parse()
     }
   }