package tutorial.storm.trident.operations;

import backtype.storm.tuple.Values;
import com.google.common.collect.ImmutableRangeMap;
import com.google.common.collect.Range;
import com.google.common.collect.RangeMap;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.tuple.TridentTuple;
import tutorial.storm.trident.testutil.Content;
import twitter4j.User;


/**
 * @author Enno Shioji (enno.shioji@peerindex.com)
 */
public class ExtractFollowerClassAndContentName extends BaseFunction {
    private RangeMap<Integer, String> discretizationMap;
    @Override
    public void execute(TridentTuple tuple, TridentCollector collector) {
        if(discretizationMap == null) discretizationMap = getDiscretizationMap();

        Content content = (Content)tuple.get(0);
        User user = (User)tuple.get(1);

        String followerClass = discretizationMap.get(user.getFollowersCount());

        collector.emit(new Values(followerClass, content.getContentName()));
    }

    private RangeMap<Integer, String> getDiscretizationMap() {
        ImmutableRangeMap.Builder<Integer,String> b = ImmutableRangeMap.builder();
        b.put(  Range.lessThan(10),                                     "< 10");
        b.put(Range.closedOpen(10,                 100),                "10-100");
        b.put(Range.closedOpen(100,               1000),                "100-1K");
        b.put(Range.closedOpen(1000,         10 * 1000),                "1K-10K");
        b.put(Range.closedOpen(10 * 1000,   100 * 1000),                "10K-100K");
        b.put(Range.closedOpen(100 * 1000, 1000 * 1000),                "100K-1M");
        b.put(Range.greaterThan(1000 * 1000),                           "> 1M");
        return b.build();
    }
}

@@ -77,143 +77,8 @@
 es;%0A
-import com.google.common.collect.ImmutableRangeMap;%0Aimport com.google.common.collect.Range;%0Aimport com.google.common.collect.RangeMap;%0A
 impo
@@ -118,16 +118,16 @@
 nction;%0A
+
 import s
@@ -415,65 +415,8 @@
 n %7B%0A
-    private RangeMap%3CInteger, String%3E discretizationMap;%0A
     
@@ -503,91 +503,8 @@
 ) %7B%0A
-        if(discretizationMap == null) discretizationMap = getDiscretizationMap();%0A%0A
     
@@ -624,29 +624,16 @@
 s = 
-discretizationMap.get
+classify
 (use
@@ -757,282 +757,81 @@
 ate 
-RangeMap%3CInteger, String%3E getDiscretizationMap() %7B%0A        ImmutableRangeMap.Builder%3CInteger,String%3E b = ImmutableRangeMap.builder();%0A        b.put(  Range.lessThan(10),                                     %22%3C 10%22);%0A        b.put(Range.closedOpen(10,                
+String classify(int followersCount) %7B%0A        if (followersCount %3C
  100)
-,   
+%7B%0A
     
@@ -842,18 +842,22 @@
     
- %2210-
+return %22%3C 
 100%22
-)
 ;%0A  
@@ -866,141 +866,46 @@
     
-b.put(Range.closedOpen(100,               1000),                %22100-1K%22);%0A        b.put(Range.closedOpen(1000,         10 * 
+%7D else if (followersCount %3C 10*
 1000)
-,  
+%7B%0A
     
@@ -916,19 +916,22 @@
     
-  %221K-
+return %22%3C 
 10K%22
-)
 ;%0A  
@@ -940,59 +940,47 @@
     
-b.put(Range.closedOpen(10 * 1000,  
+%7D else if (followersCount %3C
  100
- * 
+*
 1000)
-,   
+%7B%0A
     
@@ -991,20 +991,23 @@
     
- %2210K-
+return %22%3C 
 100K%22
-)
 ;%0A  
@@ -1016,195 +1016,62 @@
     
-b.put(Range.closedOpen(100 * 1000, 1000 * 1000),                %22100K-1M%22);%0A        b.put(Range.greaterThan(1000 * 1000),                           %22%3E 1M%22);%0A        return b.build();
+%7D else %7B%0A            return %22%3E= 100K%22;%0A        %7D
 %0A    %7D%0A
+%0A
 %7D%0A

Remove RangeMap because we have an incompatible version somewhere that breaks with "IncompatibleClass". Can't find the culprit though.