Merge 0417923 into 1c82972

README.md

@@ -66,7 +66,7 @@ In order to run the documentation locally, make sure to have [Ruby and Rubygem](
 By default the documentation will be served at [http://127.0.0.1:4000](http://127.0.0.1:4000)
 
 ## Use orchestration tools
-The orchestration tools are located under [`/tools`](/tools). They depends on:
+The orchestration tools are located under [`/tools`](/tools). They depend on:
 - Python
 - Docker
 - Redis

build.sbt

@@ -46,7 +46,8 @@ lazy val root = (project in file("."))
     pluginMongodb,
     pluginElasticSearch,
     pluginGitHub,
-    pluginRabbitMQ)
+    pluginRabbitMQ,
+    pluginGHTorrent)
 
 lazy val core = (project in file("codefeedr-core"))
   .settings(
@@ -100,9 +101,7 @@ lazy val pluginMongodb = (project in file("codefeedr-plugins/codefeedr-mongodb")
       dependencies.embeddedMongo
     )
   )
-  .dependsOn(
-    core
-  )
+  .dependsOn(core)
 
 lazy val pluginElasticSearch = (project in file("codefeedr-plugins/codefeedr-elasticsearch"))
   .settings(
@@ -113,9 +112,7 @@ lazy val pluginElasticSearch = (project in file("codefeedr-plugins/codefeedr-ela
       dependencies.flinkElasticSearch
     )
   )
-  .dependsOn(
-    core
-  )
+  .dependsOn(core)
 
 lazy val pluginGitHub = (project in file("codefeedr-plugins/codefeedr-github"))
   .settings(
@@ -130,9 +127,7 @@ lazy val pluginGitHub = (project in file("codefeedr-plugins/codefeedr-github"))
       dependencies.json4sExt
     )
   )
-  .dependsOn(
-    core
-  )
+  .dependsOn(core)
 
 lazy val pluginRabbitMQ = (project in file("codefeedr-plugins/codefeedr-rabbitmq"))
   .settings(
@@ -144,9 +139,18 @@ lazy val pluginRabbitMQ = (project in file("codefeedr-plugins/codefeedr-rabbitmq
       //dependencies.embeddedRabbitMQ
     )
   )
-  .dependsOn(core
-  )
+  .dependsOn(core)
 
+lazy val pluginGHTorrent = (project in file("codefeedr-plugins/codefeedr-ghtorrent"))
+  .settings(
+    name := pluginPrefix + "ghtorrent",
+    settings,
+    assemblySettings,
+    libraryDependencies ++= commonDependencies ++ Seq(
+      dependencies.flinkRabbitMQ,
+      //dependencies.embeddedRabbitMQ
+    )
+  ).dependsOn(core)
 
 lazy val dependencies =
   new {

codefeedr-plugins/codefeedr-ghtorrent/src/main/resources/routing_keys.txt

@@ -0,0 +1,26 @@
+evt.commitcomment.insert
+evt.create.insert
+evt.delete.insert
+evt.deployment.insert
+evt.deploymentstatus.insert
+evt.download.insert
+evt.follow.insert
+evt.fork.insert
+evt.forkapply.insert
+evt.gist.insert
+evt.gollum.insert
+evt.issuecomment.insert
+evt.issues.insert
+evt.member.insert
+evt.membership.insert
+evt.pagebuild.insert
+evt.public.insert
+evt.pullrequest.insert
+evt.pullrequestreviewcomment.insert
+evt.push.insert
+evt.release.insert
+evt.repository.insert
+evt.status.insert
+evt.teamadd.insert
+evt.watch.insert
+ent.commits.insert

codefeedr-plugins/codefeedr-ghtorrent/src/main/scala/org/codefeedr/plugins/ghtorrent/protocol/GHTorrent.scala

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.codefeedr.plugins.ghtorrent.protocol
+
+object GHTorrent {
+
+  /** Represents a record as retrieved from GHTorrent.
+    *
+    * @param routingKey the routing key (e.g. evt.push.insert)
+    * @param contents the content of the queue record.
+    */
+  case class Record(routingKey: String, contents: String)
+
+  /** Every GHTorrent record has an _id objet.
+    *
+    * @param `$oid` the object id.
+    */
+  case class _id(`$oid`: String)
+
+  /** Every event should subtype from this class. */
+  abstract class Event() extends Serializable
+}

codefeedr-plugins/codefeedr-ghtorrent/src/main/scala/org/codefeedr/plugins/ghtorrent/protocol/GitHub.scala

@@ -0,0 +1,111 @@
+package org.codefeedr.plugins.ghtorrent.protocol
+
+import java.util.Date
+
+import org.codefeedr.plugins.ghtorrent.protocol.GHTorrent.{Event, _id}
+
+object GitHub {
+
+  /**
+    * START Commit
+    */
+  case class Commit(_id: _id,
+                    node_id: String,
+                    sha: String,
+                    url: String,
+                    commit: CommitData,
+                    author: Option[User],
+                    committer: Option[User],
+                    parents: List[Parent],
+                    stats: Stats,
+                    files: List[File])
+
+  case class CommitData(author: CommitUser,
+                        committer: CommitUser,
+                        message: String,
+                        tree: Tree,
+                        comment_count: Int,
+                        verification: Verification)
+
+  case class CommitUser(name: String, email: String, date: String)
+
+  case class User(id: Long,
+                  login: String,
+                  avatar_url: String,
+                  `type`: String,
+                  site_admin: Boolean)
+
+  case class Verification(verified: Boolean,
+                          reason: String,
+                          signature: Option[String],
+                          payload: Option[String])
+
+  case class Stats(total: Int, additions: Int, deletions: Int)
+
+  case class File(sha: Option[String],
+                  filename: Option[String],
+                  status: Option[String],
+                  additions: Int,
+                  deletions: Int,
+                  changes: Int,
+                  blob_url: Option[String],
+                  raw_url: Option[String],
+                  contents_url: Option[String],
+                  patch: Option[String])
+
+  case class Parent(sha: String)
+
+  case class Tree(sha: String)
+
+  /**
+    * END Commit
+    */
+  /**
+    * START Event
+    */
+  case class Actor(id: Long,
+                   login: String,
+                   display_login: String,
+                   gravatar_id: String,
+                   url: String,
+                   avatar_url: String)
+
+  case class Repo(id: Long, name: String, url: String)
+
+  case class Organization(id: Long, login: String)
+
+  /**
+    * END Event
+    */
+  /**
+    * START PushEvent
+    */
+  case class PushEvent(id: String,
+                       _id: _id,
+                       `type`: String,
+                       actor: Actor,
+                       repo: Repo,
+                       organization: Option[Organization],
+                       payload: PushPayload,
+                       public: Boolean,
+                       created_at: Date)
+      extends Event
+
+  case class PushPayload(push_id: Long,
+                         size: Int,
+                         distinct_size: Int,
+                         ref: String,
+                         head: String,
+                         before: String,
+                         commits: List[PushCommit])
+
+  case class PushCommit(sha: String,
+                        author: PushAuthor,
+                        message: String,
+                        distinct: Boolean)
+
+  case class PushAuthor(email: String, name: String)
+  /**
+  * END PushEvent
+  */
+}

codefeedr-plugins/codefeedr-ghtorrent/src/main/scala/org/codefeedr/plugins/ghtorrent/stages/GHTInputStage.scala

@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.codefeedr.plugins.ghtorrent.stages
+
+import org.apache.flink.streaming.api.scala.DataStream
+import org.codefeedr.pipeline.Context
+import org.codefeedr.plugins.ghtorrent.protocol.GHTorrent.Record
+import org.codefeedr.plugins.ghtorrent.util.GHTorrentRabbitMQSource
+import org.codefeedr.stages.InputStage
+import org.apache.flink.api.scala._
+
+/** GHTorrent input stage following the GHTorrent Streaming specification.
+  *
+  * @param username your username to specify to GHTorrent.
+  * @param stageName name of this stage, "ghtorrent_input" is default.
+  * @param host the host name, default is localhost.
+  * @param port the port, default is 5672.
+  * @param routingKeysFile the location of the routingKeysFile (in the resources directory).
+  */
+class GHTInputStage(username: String,
+                    stageName: String = "ght_input",
+                    host: String = "localhost",
+                    port: Int = 5672,
+                    routingKeysFile: String = "routing_keys.txt")
+    extends InputStage[Record](Some(stageName)) {
+
+  /** Transforms GHTorrent data to a Record format.
+    *
+    * @param context the CodeFeedr context.
+    * @return A newly created DataStream of Records.
+    */
+  override def main(context: Context): DataStream[Record] = {
+    context.env
+      .addSource(
+        new GHTorrentRabbitMQSource(username, host, port, routingKeysFile))
+      .map { x =>
+        val splitEl = x.split("#", 2)
+        val routingKey = splitEl(0)
+        val record = splitEl(1)
+
+        Record(routingKey, record)
+      }
+  }
+}

codefeedr-plugins/codefeedr-ghtorrent/src/main/scala/org/codefeedr/plugins/ghtorrent/stages/GHTRecordToCommitStage.scala

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.codefeedr.plugins.ghtorrent.stages
+
+import org.apache.flink.streaming.api.scala.DataStream
+import org.codefeedr.plugins.ghtorrent.protocol.GHTorrent.Record
+import org.codefeedr.plugins.ghtorrent.protocol.GitHub.Commit
+import org.codefeedr.stages.TransformStage
+import org.json4s.DefaultFormats
+import org.json4s.ext.JavaTimeSerializers
+import org.json4s.jackson.JsonMethods.parse
+import org.apache.flink.api.scala._
+
+class GHTRecordToCommitStage(stageName: String = "commits")
+    extends TransformStage[Record, Commit](Some(stageName)) {
+
+  override def transform(source: DataStream[Record]): DataStream[Commit] = {
+    source
+      .filter(_.routingKey == "ent.commits.insert")
+      .map { x =>
+        implicit val defaultFormats = DefaultFormats ++ JavaTimeSerializers.all
+        parse(x.contents).extract[Commit]
+      }
+  }
+}

codefeedr-plugins/codefeedr-ghtorrent/src/main/scala/org/codefeedr/plugins/ghtorrent/stages/GHTRecordToEventStage.scala

@@ -0,0 +1,41 @@
+package org.codefeedr.plugins.ghtorrent.stages
+
+import org.apache.flink.api.common.functions.{FilterFunction, MapFunction}
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.streaming.api.scala.DataStream
+import org.codefeedr.plugins.ghtorrent.protocol.GHTorrent.{Event, Record}
+import org.codefeedr.stages.TransformStage
+import org.json4s.DefaultFormats
+import org.json4s.ext.JavaTimeSerializers
+import org.json4s.jackson.JsonMethods.parse
+import org.apache.flink.api.scala._
+import org.codefeedr.plugins.ghtorrent.protocol.GitHub.PushEvent
+
+import scala.reflect.{ClassTag, classTag}
+import scala.reflect.runtime.universe._
+
+class GHTRecordToEventStage[
+    T <: Serializable with AnyRef with Event: TypeTag: ClassTag: TypeInformation](
+    stageName: String,
+    var routingKey: String)
+    extends TransformStage[Record, T] {
+
+  override def transform(source: DataStream[Record]): DataStream[T] = {
+    source
+      .filter(new RoutingFilter(routingKey))
+      .map(new EventExtract[T])
+  }
+
+  def getKey() = routingKey
+}
+
+class RoutingFilter(routingKey: String) extends FilterFunction[Record] {
+  override def filter(value: Record): Boolean = value.routingKey == routingKey
+}
+
+class EventExtract[T: Manifest] extends MapFunction[Record, T] {
+  override def map(value: Record): T = {
+    implicit val defaultFormats = DefaultFormats ++ JavaTimeSerializers.all
+    parse(value.contents).extract[T]
+  }
+}