Make EdgeDirection val instead of case object for Java compat.

3 files changed, 15 insertions, 5 deletions

diff --git a/graphx/src/main/scala/org/apache/spark/graphx/EdgeDirection.scala b/graphx/src/main/scala/org/apache/spark/graphx/EdgeDirection.scala
index 05103f6883..9d37f6513f 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/EdgeDirection.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/EdgeDirection.scala
@@ -3,7 +3,7 @@ package org.apache.spark.graphx
 /**
  * The direction of a directed edge relative to a vertex.
  */
-sealed abstract class EdgeDirection {
+class EdgeDirection private (private val name: String) extends Serializable {
   /**
    * Reverse the direction of an edge.  An in becomes out,
    * out becomes in and both remains both.
@@ -13,16 +13,25 @@ sealed abstract class EdgeDirection {
     case EdgeDirection.Out  => EdgeDirection.In
     case EdgeDirection.Both => EdgeDirection.Both
   }
+
+  override def toString: String = "EdgeDirection." + name
+
+  override def equals(o: Any) = o match {
+    case other: EdgeDirection => other.name == name
+    case _ => false
+  }
+
+  override def hashCode = name.hashCode
 }
 
 
 object EdgeDirection {
   /** Edges arriving at a vertex. */
-  case object In extends EdgeDirection
+  final val In = new EdgeDirection("In")
 
   /** Edges originating from a vertex. */
-  case object Out extends EdgeDirection
+  final val Out = new EdgeDirection("Out")
 
   /** All edges adjacent to a vertex. */
-  case object Both extends EdgeDirection
+  final val Both = new EdgeDirection("Both")
 }
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/GraphKryoRegistrator.scala b/graphx/src/main/scala/org/apache/spark/graphx/GraphKryoRegistrator.scala
index 681074ba10..d79bdf9618 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/GraphKryoRegistrator.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/GraphKryoRegistrator.scala
@@ -23,6 +23,7 @@ class GraphKryoRegistrator extends KryoRegistrator {
     kryo.register(classOf[VertexAttributeBlock[Object]])
     kryo.register(classOf[PartitionStrategy])
     kryo.register(classOf[BoundedPriorityQueue[Object]])
+    kryo.register(classOf[EdgeDirection])
 
     // This avoids a large number of hash table lookups.
     kryo.setReferences(false)
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/impl/GraphImpl.scala b/graphx/src/main/scala/org/apache/spark/graphx/impl/GraphImpl.scala
index c66b8c804f..6a2abc71cc 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/impl/GraphImpl.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/impl/GraphImpl.scala
@@ -283,7 +283,7 @@ class GraphImpl[VD: ClassTag, ED: ClassTag] protected (
           }
         case Some(EdgeDirection.In) =>
           edgePartition.iterator.filter(e => vPart.isActive(e.dstId))
-        case None =>
+        case _ => // None
           edgePartition.iterator
       }