//设置package包名称以及导入依赖的类
package com.init6.users

import java.util.Comparator

import akka.actor.ActorSystem
import akka.dispatch.{Envelope, UnboundedStablePriorityMailbox}
import com.typesafe.config.Config


class UsersMailbox(settings: ActorSystem.Settings, config: Config)
  extends UnboundedStablePriorityMailbox(UsersPriorityGenerator)

object UsersPriorityGenerator extends Comparator[Envelope] {

  override def compare(o1: Envelope, o2: Envelope) = {
    if (o1.message.isInstanceOf[Add] && o2.message.isInstanceOf[Add]) {
      val m1 = o1.message.asInstanceOf[Add]
      val m2 = o2.message.asInstanceOf[Add]

      if (m1.connectionInfo.connectedTime > m2.connectionInfo.connectedTime) {
        1
      } else {
        -1
      }
    } else {
      0
    }
  }
} 

//设置package包名称以及导入依赖的类
package pl.writeonly.babel.entities

import java.util.Comparator
import pl.writeonly.scala.util.ToBoolean
import javax.jdo.annotations.Join
import javax.jdo.annotations.Persistent

trait Evaluator[T] {
  def evaluate(paramT1: T, paramT2: T): Int;

  def equals(paramObject: AnyRef): Boolean;
}

object Record extends Comparator[Record] with ToBoolean {
  var comparator: Comparator[Record] = _
  var evaluator: Evaluator[Record] = _

  def compare(key: Record, value: Record): Int = {
    if (comparator != null) {
      //return comparator(key, value)
      return 0;
    }
    else {
      return evaluate(key) - evaluate(value)
    }
  }
  def evaluate(record: Record): Int = {
    if (evaluator != null) {
      return 0
    }
    else {
      return record.corect - record.wrong
    }
  }
  def int(string: String) = new Integer(string) toInt
}

class Record(@Persistent val user: User, @Persistent val relation: Relation, var corect: Int, var wrong: Int)
  extends Entity {
  def this(user: User, relation: Relation) = this(user, relation, 0, 0)
  def this(strings: Array[String]) = this(new User(strings(0)), new Relation(strings(1)), Record.int(strings(2)), Record.int(strings(3)))
  def this(string: String) = this(string split ",")
  def ++ = corect + 1
  def -- = wrong + 1
  def eval = Record evaluate this
  override def toString = "" + user + "," + relation + "," + corect + "," + wrong

} 

//设置package包名称以及导入依赖的类
package net.sansa_stack.inference.flink

import java.util
import java.util.Comparator

import com.google.common.collect.ComparisonChain
import net.sansa_stack.inference.flink.data.RDFGraph
import org.apache.flink.api.scala.{ExecutionEnvironment, _}
import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
import org.apache.flink.test.util.{MultipleProgramsTestBase, TestBaseUtils}
import org.junit.Test
import org.junit.runner.RunWith
import org.junit.runners.Parameterized
import net.sansa_stack.inference.data.RDFTriple

import scala.collection.JavaConverters._


@RunWith(classOf[Parameterized])
class RDFGraphTestCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {

  @Test
  def testSubtract(): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment


    // generate dataset
    val g1 = RDFGraph(env.fromCollection(
      Seq(
        RDFTriple("s1", "p1", "o1"),
        RDFTriple("s1", "p1", "o2"),
        RDFTriple("s1", "p1", "o3")
      )
    ))
    val g2 = RDFGraph(env.fromCollection(
      Seq(
        RDFTriple("s1", "p1", "o1"),
        RDFTriple("s1", "p1", "o2")
      )
    ))

    // compute
    val g_diff = g1.subtract(g2)

    val result = g_diff.triples.collect()
    val expected = Seq(
      RDFTriple("s1", "p1", "o3")
    )

    TestBaseUtils.compareResultCollections(new util.ArrayList(result.asJava), new util.ArrayList(expected.asJava), new Comparator[RDFTriple] {
      override def compare(t1: RDFTriple, t2: RDFTriple): Int =
        ComparisonChain.start()
          .compare(t1.s, t2.s)
          .compare(t1.p, t2.p)
          .compare(t1.o, t2.o)
        .result()
    })
  }

}