BroadcastMatcher modified to use HadoopMapReader

BroadcastMatcher.scala

package com.admiral.barefoot_spark

import java.net.URI
import java.util.ArrayList
import java.util.{List => JavaList}
import java.util.Properties

import scala.collection.JavaConverters._

import com.bmwcarit.barefoot.matcher.{Matcher, MatcherKState, MatcherSample}
import com.bmwcarit.barefoot.roadmap.{Road, RoadMap, RoadPoint, TimePriority}
import com.bmwcarit.barefoot.spatial.Geography
import com.bmwcarit.barefoot.topology.Dijkstra
import com.esri.core.geometry.Point


object BroadcastMatcher {
  private var instance = null: Matcher

  private def initialize(map_uri: String) {
    // Barefoot page suggests code that uses a double-checked lock. See
    // https://www.cs.umd.edu/~pugh/java/memoryModel/DoubleCheckedLocking.html
    // for why this was removed here.
    this.synchronized {
      if (instance == null) {
        // initialize map matcher once per Executor (JVM process/cluster node)
        val map = RoadMap.Load(new HadoopMapReader(new URI(map_uri))).construct()
        val router = new Dijkstra[Road, RoadPoint]()
        val cost = new TimePriority()
        val spatial = new Geography()

        instance = new Matcher(map, router, cost, spatial)
      }
    }
  }
}

@SerialVersionUID(1L)
class BroadcastMatcher(map_uri: String) extends Serializable {

  def mmatch(samples: JavaList[MatcherSample]): MatcherKState = {
    mmatch(samples, 0, 0)
  }

  def mmatch(samples: JavaList[MatcherSample], minDistance: Double, minInterval: Int): MatcherKState = {
    BroadcastMatcher.initialize(map_uri)
    BroadcastMatcher.instance.mmatch(samples, minDistance, minInterval)
  }
}