<commit_before>package org.calrissian.flowbox.bolt;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.StormTopology;
import org.calrissian.flowbox.FlowboxTopology;
import org.calrissian.flowbox.model.Event;
import org.calrissian.flowbox.model.Flow;
import org.calrissian.flowbox.model.Policy;
import org.calrissian.flowbox.model.Tuple;
import org.calrissian.flowbox.model.builder.FlowBuilder;
import org.calrissian.flowbox.spout.MockEventGeneratorSpout;
import org.calrissian.flowbox.spout.MockFlowLoaderSpout;
import org.calrissian.flowbox.support.Aggregator;
import org.calrissian.flowbox.support.WindowItem;
import org.junit.Before;
import org.junit.Test;

import java.util.Collections;
import java.util.List;
import java.util.UUID;

import static java.lang.System.currentTimeMillis;
import static java.util.Collections.singletonList;
import static org.junit.Assert.assertEquals;

public class AggregatorBoltIT {

    @Before
    public void setUp() {

    }

    private StormTopology buildTopology(Flow flow) {
        StormTopology topology = new FlowboxTopology().buildTopology(
                new MockFlowLoaderSpout(singletonList(flow), 60000),
                new MockEventGeneratorSpout(10),
                new MockSinkBolt(),
                6);

        return topology;
    }

    @Test
    public void test() {

        Flow flow = new FlowBuilder()
            .id("myflow")
            .flowDefs()
                .stream("stream1")
                    .partition().field("key3").end()
                    .aggregate().aggregator(TestAggregator.class).trigger(Policy.TIME, 5).evict(Policy.TIME, 10).end()
                .endStream()
            .endDefs()
        .createFlow();

        StormTopology topology = buildTopology(flow);
        Config conf = new Config();
        conf.setNumWorkers(20);

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("test", conf, topology);

        try {
            Thread.sleep(25000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        assertEquals(4, MockSinkBolt.getEvents().size());
    }

    public static final class TestAggregator implements Aggregator {

        private long count;

        @Override
        public void added(WindowItem item) {
            count++;
        }

        @Override
        public void evicted(WindowItem item) {
            count--;

        }

        @Override
        public List<Event> aggregate() {
            Event toReturn = new Event(UUID.randomUUID().toString(), currentTimeMillis());
            toReturn.put(new Tuple("count", count));

            count = 0;
            return singletonList(toReturn);
        }

    }
}
<commit_msg>Aggregator bolt should be working<commit_after>package org.calrissian.flowbox.bolt;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.StormTopology;
import org.calrissian.flowbox.FlowboxTopology;
import org.calrissian.flowbox.model.Event;
import org.calrissian.flowbox.model.Flow;
import org.calrissian.flowbox.model.Policy;
import org.calrissian.flowbox.model.Tuple;
import org.calrissian.flowbox.model.builder.FlowBuilder;
import org.calrissian.flowbox.spout.MockEventGeneratorSpout;
import org.calrissian.flowbox.spout.MockFlowLoaderSpout;
import org.calrissian.flowbox.support.Aggregator;
import org.calrissian.flowbox.support.WindowItem;
import org.junit.Before;
import org.junit.Test;

import java.util.List;
import java.util.UUID;

import static java.lang.System.currentTimeMillis;
import static java.util.Collections.singletonList;
import static org.junit.Assert.assertEquals;

public class AggregatorBoltIT {

    private StormTopology buildTopology(Flow flow) {
        StormTopology topology = new FlowboxTopology().buildTopology(
                new MockFlowLoaderSpout(singletonList(flow), 60000),
                new MockEventGeneratorSpout(10),
                new MockSinkBolt(),
                6);

        return topology;
    }

    @Test
    public void test() {

        Flow flow = new FlowBuilder()
            .id("myflow")
            .flowDefs()
                .stream("stream1")
                    .partition().field("key3").end()
                    .aggregate().aggregator(TestAggregator.class).trigger(Policy.TIME, 5).evict(Policy.TIME, 10).end()
                .endStream()
            .endDefs()
        .createFlow();

        StormTopology topology = buildTopology(flow);
        Config conf = new Config();
        conf.setNumWorkers(20);

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("test", conf, topology);

        try {
            Thread.sleep(25000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        assertEquals(4, MockSinkBolt.getEvents().size());
    }

    public static final class TestAggregator implements Aggregator {

        private long count;

        @Override
        public void added(WindowItem item) {
            count++;
        }

        @Override
        public void evicted(WindowItem item) {
            count--;

        }

        @Override
        public List<Event> aggregate() {
            Event toReturn = new Event(UUID.randomUUID().toString(), currentTimeMillis());
            toReturn.put(new Tuple("count", count));

            count = 0;
            return singletonList(toReturn);
        }

    }
}

{'fraction_non_alphanumeric': 0.10346083788706739, 'fraction_numerical': 0.008014571948998178, 'mean_word_length': 2.5176169122357464, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7486924', 'n_tokens_mistral': 1613, 'n_tokens_neox': 1549, 'n_words': 300}