Merge pull request #8 from traveltime-dev/time-filter-support

Time filter support

8/src/main/java/com/traveltime/plugin/solr/TimeFilterQParserPlugin.java

@@ -0,0 +1,44 @@
+package com.traveltime.plugin.solr;
+
+import com.traveltime.plugin.solr.cache.RequestCache;
+import com.traveltime.plugin.solr.fetcher.JsonFetcherSingleton;
+import com.traveltime.plugin.solr.query.timefilter.TimeFilterQueryParser;
+import org.apache.solr.common.params.SolrParams;
+import org.apache.solr.common.util.NamedList;
+import org.apache.solr.request.SolrQueryRequest;
+import org.apache.solr.search.QParser;
+import org.apache.solr.search.QParserPlugin;
+
+import java.net.URI;
+import java.util.Optional;
+
+public class TimeFilterQParserPlugin extends QParserPlugin {
+   private String cacheName = RequestCache.NAME;
+
+   private static final Integer DEFAULT_LOCATION_SIZE_LIMIT = 2000;
+
+   @Override
+   public void init(NamedList args) {
+      Object cache = args.get("cache");
+      if(cache != null) cacheName = cache.toString();
+
+      Object uriVal = args.get("api_uri");
+      URI uri = null;
+      if(uriVal != null) uri = URI.create(uriVal.toString());
+
+      String appId = args.get("app_id").toString();
+      String apiKey = args.get("api_key").toString();
+      int locationLimit =
+              Optional.ofNullable(args.get("location_limit"))
+                      .map(x -> Integer.parseInt(x.toString()))
+                      .orElse(DEFAULT_LOCATION_SIZE_LIMIT);
+
+      JsonFetcherSingleton.INSTANCE.init(uri, appId, apiKey, locationLimit);
+   }
+
+   @Override
+   public QParser createParser(String qstr, SolrParams localParams, SolrParams params, SolrQueryRequest req) {
+      return new TimeFilterQueryParser(qstr, localParams, params, req, JsonFetcherSingleton.INSTANCE.getFetcher(), cacheName);
+   }
+
+}

8/src/main/java/com/traveltime/plugin/solr/TraveltimeQParserPlugin.java

@@ -1,6 +1,7 @@
 package com.traveltime.plugin.solr;
 
 import com.traveltime.plugin.solr.cache.RequestCache;
+import com.traveltime.plugin.solr.fetcher.ProtoFetcherSingleton;
 import com.traveltime.plugin.solr.query.TraveltimeQueryParser;
 import org.apache.solr.common.params.SolrParams;
 import org.apache.solr.common.util.NamedList;
@@ -11,7 +12,6 @@
 import java.net.URI;
 
 public class TraveltimeQParserPlugin extends QParserPlugin {
-   public static String PARAM_PREFIX = "traveltime_";
    private String cacheName = RequestCache.NAME;
 
    @Override
@@ -25,12 +25,12 @@ public void init(NamedList args) {
 
       String appId = args.get("app_id").toString();
       String apiKey = args.get("api_key").toString();
-      FetcherSingleton.INSTANCE.init(uri, appId, apiKey);
+      ProtoFetcherSingleton.INSTANCE.init(uri, appId, apiKey);
    }
 
    @Override
    public QParser createParser(String qstr, SolrParams localParams, SolrParams params, SolrQueryRequest req) {
-      return new TraveltimeQueryParser(qstr, localParams, params, req, FetcherSingleton.INSTANCE.getFetcher(), cacheName);
+      return new TraveltimeQueryParser(qstr, localParams, params, req, ProtoFetcherSingleton.INSTANCE.getFetcher(), cacheName);
    }
 
 }

8/src/main/java/com/traveltime/plugin/solr/cache/ExactRequestCache.java

@@ -2,7 +2,7 @@
 
 import com.traveltime.plugin.solr.query.TraveltimeQueryParameters;
 
-public class ExactRequestCache extends RequestCache {
+public class ExactRequestCache extends RequestCache<TraveltimeQueryParameters> {
    private final Object[] lock = new Object[0];
 
    @Override

8/src/main/java/com/traveltime/plugin/solr/cache/ExactTimeFilterRequestCache.java

@@ -0,0 +1,22 @@
+package com.traveltime.plugin.solr.cache;
+
+import com.traveltime.plugin.solr.query.timefilter.TimeFilterQueryParameters;
+
+public class ExactTimeFilterRequestCache extends RequestCache<TimeFilterQueryParameters> {
+   private final Object[] lock = new Object[0];
+
+   @Override
+   public TravelTimes getOrFresh(TimeFilterQueryParameters key) {
+      TravelTimes result = get(key);
+      if (result == null) {
+         synchronized (lock) {
+            result = get(key);
+            if (result == null) {
+               result = new BasicTravelTimes();
+               put(key, result);
+            }
+         }
+      }
+      return result;
+   }
+}

8/src/main/java/com/traveltime/plugin/solr/cache/FuzzyRequestCache.java

@@ -5,7 +5,7 @@
 
 import java.util.Map;
 
-public class FuzzyRequestCache extends RequestCache {
+public class FuzzyRequestCache extends RequestCache<TraveltimeQueryParameters> {
    private final Object[] lock = new Object[0];
    private Map<String, String> args;
 

8/src/main/java/com/traveltime/plugin/solr/cache/FuzzyTimeFilterRequestCache.java

@@ -0,0 +1,33 @@
+package com.traveltime.plugin.solr.cache;
+
+import com.traveltime.plugin.solr.query.timefilter.TimeFilterQueryParameters;
+import org.apache.solr.search.CacheRegenerator;
+
+import java.util.Map;
+
+public class FuzzyTimeFilterRequestCache extends RequestCache<TimeFilterQueryParameters> {
+   private final Object[] lock = new Object[0];
+   private Map<String, String> args;
+
+   @Override
+   public Object init(Map args, Object persistence, CacheRegenerator regenerator) {
+      this.args = args;
+      return super.init(args, persistence, regenerator);
+   }
+
+   @Override
+   public TravelTimes getOrFresh(TimeFilterQueryParameters key) {
+      key = key.withField(null).withTravelTime(0);
+      TravelTimes result = get(key);
+      if (result == null) {
+         synchronized (lock) {
+            result = get(key);
+            if (result == null) {
+               result = new LRUTimes(args);
+               put(key, result);
+            }
+         }
+      }
+      return result;
+   }
+}

8/src/main/java/com/traveltime/plugin/solr/cache/RequestCache.java

@@ -7,10 +7,10 @@
 
 import java.util.Map;
 
-public abstract class RequestCache extends FastLRUCache<TraveltimeQueryParameters, TravelTimes> {
+public abstract class RequestCache<P> extends FastLRUCache<P, TravelTimes> {
    public static String NAME = "traveltime";
 
-   public abstract TravelTimes getOrFresh(TraveltimeQueryParameters key);
+   public abstract TravelTimes getOrFresh(P key);
 
    @Override
    public void init(Map<String, String> args, CacheRegenerator ignored) {

8/src/main/java/com/traveltime/plugin/solr/fetcher/Fetcher.java

@@ -0,0 +1,10 @@
+package com.traveltime.plugin.solr.fetcher;
+
+import com.traveltime.sdk.dto.common.Coordinates;
+import java.util.ArrayList;
+
+import java.util.List;
+
+public interface Fetcher<Params> {
+    List<Integer> getTimes(Params parameters, ArrayList<Coordinates> points);
+}

8/src/main/java/com/traveltime/plugin/solr/fetcher/JsonFetcher.java

@@ -0,0 +1,158 @@
+package com.traveltime.plugin.solr.fetcher;
+
+import com.google.common.collect.Iterables;
+import com.traveltime.plugin.solr.query.timefilter.TimeFilterQueryParameters;
+import com.traveltime.plugin.solr.util.Util;
+import com.traveltime.sdk.TravelTimeSDK;
+import com.traveltime.sdk.auth.TravelTimeCredentials;
+import com.traveltime.sdk.dto.common.Coordinates;
+import com.traveltime.sdk.dto.common.Location;
+import com.traveltime.sdk.dto.common.Property;
+import com.traveltime.sdk.dto.requests.TimeFilterRequest;
+import com.traveltime.sdk.dto.requests.timefilter.ArrivalSearch;
+import com.traveltime.sdk.dto.requests.timefilter.DepartureSearch;
+import com.traveltime.sdk.dto.responses.TimeFilterResponse;
+import com.traveltime.sdk.dto.responses.errors.IOError;
+import com.traveltime.sdk.dto.responses.errors.ResponseError;
+import com.traveltime.sdk.dto.responses.errors.TravelTimeError;
+import lombok.val;
+import okhttp3.OkHttpClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.StreamSupport;
+
+public class JsonFetcher implements Fetcher<TimeFilterQueryParameters> {
+   private final TravelTimeSDK api;
+
+   private final int locationSizeLimit;
+
+   private final Logger log = LoggerFactory.getLogger(JsonFetcher.class);
+
+   private void logError(TravelTimeError left) {
+      if (left instanceof IOError) {
+         val ioerr = (IOError) left;
+         log.warn(ioerr.getMessage());
+         log.warn(
+                 Arrays.stream(ioerr.getCause().getStackTrace())
+                         .map(StackTraceElement::toString)
+                         .reduce("", (a, b) -> a + "\n\t" + b)
+         );
+      } else if (left instanceof ResponseError) {
+         val error = (ResponseError) left;
+         log.warn(error.getDescription());
+      }
+   }
+
+   public JsonFetcher(URI uri, String id, String key, int locationSizeLimit) {
+      val auth = TravelTimeCredentials.builder().appId(id).apiKey(key).build();
+      val client = new OkHttpClient.Builder()
+              .connectTimeout(5, TimeUnit.MINUTES)
+              .callTimeout(5, TimeUnit.MINUTES)
+              .readTimeout(5, TimeUnit.MINUTES)
+              .build();
+      val builder = TravelTimeSDK.builder().credentials(auth).client(client);
+      if(uri != null) {
+         builder.baseProtoUri(uri);
+      }
+      api = builder.build();
+      this.locationSizeLimit = locationSizeLimit;
+   }
+
+   private Integer[] extractTimes(TimeFilterResponse response, Integer[] travelTimes) {
+      val result = response.getResults().get(0);
+
+      result.getLocations()
+              .forEach(location ->
+                      travelTimes[Integer.parseInt(location.getId())] = location.getProperties().get(0).getTravelTime()
+              );
+
+      result.getUnreachable()
+              .forEach(unreachableId ->
+                      travelTimes[Integer.parseInt(unreachableId)] = -1
+              );
+
+      return travelTimes;
+   }
+
+   public List<Integer> getTimes(TimeFilterQueryParameters parameters, ArrayList<Coordinates> points) {
+
+      val locations = IntStream
+              .range(0, points.size())
+              .mapToObj(i -> new Location(String.valueOf(i), points.get(i)))
+              .collect(Collectors.toList());
+
+      val groupedLocations = Iterables.partition(locations, locationSizeLimit);
+
+      val requests = StreamSupport.stream(groupedLocations.spliterator(), true)
+              .map(locationGroup -> {
+                 val requestBuilder = TimeFilterRequest.builder();
+
+                 requestBuilder
+                         .location(parameters.getLocation())
+                         .locations(locations);
+
+                  switch (parameters.getSearchType()) {
+                      case ARRIVAL:
+                          val arrivalSearchBuilder = ArrivalSearch
+                                  .builder()
+                                  .id("search")
+                                  .arrivalLocationId(parameters.getLocation().getId())
+                                  .departureLocationIds(locations.stream().map(Location::getId).collect(Collectors.toList()))
+                                  .arrivalTime(parameters.getTime())
+                                  .travelTime(parameters.getTravelTime())
+                                  .properties(Collections.singletonList(Property.TRAVEL_TIME))
+                                  .transportation(parameters.getTransportation());
+                          val arrivalSearch = parameters
+                                  .getRange()
+                                  .map(arrivalSearchBuilder::range)
+                                  .orElse(arrivalSearchBuilder)
+                                  .build();
+                          requestBuilder.arrivalSearch(arrivalSearch);
+                          break;
+                      case DEPARTURE:
+                          val departureSearchBuilder = DepartureSearch
+                                  .builder()
+                                  .id("search")
+                                  .departureLocationId(parameters.getLocation().getId())
+                                  .arrivalLocationIds(locations.stream().map(Location::getId).collect(Collectors.toList()))
+                                  .departureTime(parameters.getTime())
+                                  .travelTime(parameters.getTravelTime())
+                                  .properties(Collections.singletonList(Property.TRAVEL_TIME))
+                                  .transportation(parameters.getTransportation());
+                          val departureSearch = parameters
+                                  .getRange()
+                                  .map(departureSearchBuilder::range)
+                                  .orElse(departureSearchBuilder)
+                                  .build();
+                          requestBuilder.departureSearch(departureSearch);
+                          break;
+                  }
+
+                 return requestBuilder.build();
+              });
+
+       log.info(String.format("Fetching %d locations", points.size()));
+
+       Integer[] resultArray = new Integer[locations.size()];
+
+       requests.map(request -> Util.time(log, () -> api.send(request)))
+               .forEach(result -> result.fold(err -> {
+                                   logError(err);
+                                   throw new RuntimeException(err.toString());
+                               },
+                               succ -> extractTimes(succ, resultArray))
+               );
+
+       return Arrays.stream(resultArray).collect(Collectors.toList());
+   }
+
+}

8/src/main/java/com/traveltime/plugin/solr/fetcher/JsonFetcherSingleton.java

@@ -0,0 +1,23 @@
+package com.traveltime.plugin.solr.fetcher;
+
+import java.net.URI;
+
+public enum JsonFetcherSingleton {
+   INSTANCE;
+
+   private JsonFetcher underlying = null;
+   private final Object[] lock = new Object[0];
+
+   public void init(URI uri, String id, String key, int locationSizeLimit) {
+      if(underlying != null) return;
+      synchronized (lock) {
+         if(underlying != null) return;
+         underlying = new JsonFetcher(uri, id, key, locationSizeLimit);
+      }
+   }
+
+   public JsonFetcher getFetcher() {
+      return underlying;
+   }
+
+}