challenge redislabs-training#5

Author: victoriaum

src/main/java/com/redislabs/university/RU102J/dao/SiteGeoDaoRedisImpl.java

@@ -52,10 +52,49 @@ public Set<Site> findByGeo(GeoQuery query) {
         }
     }
 
+    private Set<Site> findSitesByGeoWithCapacity(GeoQuery query) {
+        Set<Site> results = new HashSet<>();
+        Coordinate coord = query.getCoordinate();
+        Double radius = query.getRadius();
+        GeoUnit radiusUnit = query.getRadiusUnit();
+
+        try (Jedis jedis = jedisPool.getResource()) {
+            // START Challenge #5
+            List<GeoRadiusResponse> radiusResponses =
+                    jedis.georadius(RedisSchema.getSiteGeoKey(), coord.getLng(),
+                            coord.getLat(), radius, radiusUnit);
+            // END Challenge #5
+
+            Set<Site> sites = radiusResponses.stream()
+                    .map(response -> jedis.hgetAll(response.getMemberByString()))
+                    .filter(Objects::nonNull)
+                    .map(Site::new).collect(Collectors.toSet());
+
+            // START Challenge #5
+            Pipeline pipeline = jedis.pipelined();
+            Map<Long, Response<Double>> scores = new HashMap<>(sites.size());
+            for (Site site : sites) {
+                Response<Double> score = pipeline.zscore(RedisSchema.getCapacityRankingKey(),
+                        String.valueOf(site.getId()));
+                scores.put(site.getId(), score);
+            }
+            pipeline.sync();
+            // END Challenge #5
+
+            for (Site site : sites) {
+                if (scores.get(site.getId()).get() >= capacityThreshold) {
+                    results.add(site);
+                }
+            }
+        }
+
+        return results;
+    }
+
     // Challenge #5
-     private Set<Site> findSitesByGeoWithCapacity(GeoQuery query) {
-         return Collections.emptySet();
-     }
+//  private Set<Site> findSitesByGeoWithCapacity(GeoQuery query) {
+//       return Collections.emptySet();
+//  }
     // Comment out the above, and uncomment what's below
 //    private Set<Site> findSitesByGeoWithCapacity(GeoQuery query) {
 //        Set<Site> results = new HashSet<>();