Test Page History Fixed Point algorithm.scala

val knownStates = sc.parallelize(Array.empty[PageState])

val a2b = new PageEvent( oldTitle = Some("A"), newTitle = Some("B"), timestamp = Some("2011"),
    eventType = Some("move"), causedByUserId = Some(1), causedByUserName = Some("Dan")
)
val b2c = new PageEvent( oldTitle = Some("B"), newTitle = Some("C"), timestamp = Some("2015"),
    eventType = Some("move"), causedByUserId = Some(1), causedByUserName = Some("Dan")
)
val d2e = new PageEvent( oldTitle = Some("D"), newTitle = Some("E"), timestamp = Some("2014"),
    eventType = Some("move"), causedByUserId = Some(1), causedByUserName = Some("Dan")
)
val c = new PageState( title = Some("C"),
    pageId = Some(1), pageCreationTimestamp = Some("2010"),
    causedByEventType = Some("create"),
    causedByUserId = Some(1),
    causedByUserName = Some("Dan")
)
val b = new PageState( title = Some("B"),
    pageId = Some(3), pageCreationTimestamp = Some("2015"),
    causedByEventType = Some("create"),
    causedByUserId = Some(1),
    causedByUserName = Some("Dan")
)
val a = new PageState( title = Some("A"),
    pageId = Some(2), pageCreationTimestamp = Some("2011"),
    causedByEventType = Some("create"),
    causedByUserId = Some(1),
    causedByUserName = Some("Dan")
)
val e = new PageState( title = Some("E"),
    pageId = Some(4), pageCreationTimestamp = Some("2012"),
    causedByEventType = Some("create"),
    causedByUserId = Some(1),
    causedByUserName = Some("Dan")
)
val d = new PageState( title = Some("D"),
    pageId = Some(5), pageCreationTimestamp = Some("2014"),
    causedByEventType = Some("create"),
    causedByUserId = Some(1),
    causedByUserName = Some("Dan")
)
val todo = sc.parallelize(Array(a2b, b2c, d2e))
val doing = sc.parallelize(Array(a, b, c, d, e))
val done = sc.parallelize(Array.empty[PageState])
fixedPoint(todo, doing, done).take(10)