@@ -0,0 +1,2458 @@
+package com.rackspacecloud.blueflood.service;%0A%0Aimport com.rackspacecloud.blueflood.service.Configuration;%0Aimport com.rackspacecloud.blueflood.service.CoreConfig;%0Aimport com.rackspacecloud.blueflood.service.ScheduleContext;%0Aimport com.rackspacecloud.blueflood.service.ShardStatePuller;%0Aimport com.rackspacecloud.blueflood.service.ShardStatePusher;%0Aimport com.rackspacecloud.blueflood.utils.Util;%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0Aimport java.util.Collection;%0Aimport java.util.Collections;%0A%0A/**%0A * Aggregates the ShardStateServices.%0A * todo: consider moving this to blueflood-core.%0A */%0Apublic class ShardStateServices %7B%0A    private static final Logger log = LoggerFactory.getLogger(ShardStateServices.class);%0A%0A    private final ScheduleContext context;%0A    private final ShardStatePusher pusher;%0A    private final ShardStatePuller puller;%0A%0A    public ShardStateServices(ScheduleContext context) %7B%0A        this.context = context;%0A%0A        // these threads are responsible for sending/receiving schedule context state to/from the database.%0A        final Collection%3CInteger%3E allShards = Collections.unmodifiableCollection(Util.parseShards(%22ALL%22));%0A        pusher = new ShardStatePusher(allShards, context.getShardStateManager());%0A        puller = new ShardStatePuller(allShards, context.getShardStateManager());%0A%0A        pusher.setActive(false);%0A        puller.setActive(false);%0A%0A       Configuration config = Configuration.getInstance();%0A        if (config.getBooleanProperty(CoreConfig.INGEST_MODE) %7C%7C config.getBooleanProperty(CoreConfig.ROLLUP_MODE)) %7B%0A            try %7B%0A                final Thread shardPush = new Thread(pusher, %22Shard state writer%22);%0A                final Thread shardPull = new Thread(puller, %22Shard state reader%22);%0A%0A                shardPull.start();%0A                shardPush.start();%0A%0A                log.info(%22Shard push and pull services started%22);%0A            %7D catch (NumberFormatException ex) %7B%0A                log.error(%22Shard services not started. Probably misconfiguration%22, ex);%0A            %7D%0A        %7D else %7B%0A            log.info(%22Shard push and pull services not required%22);%0A        %7D%0A%0A        // if things were enabled, threads are actually running at this point, but are blocked until enabled.%0A    %7D%0A%0A    public void start() %7B%0A        pusher.setActive(true);%0A        pusher.setActive(false);%0A    %7D%0A%0A    public void stop() %7B%0A        pusher.setActive(false);%0A        puller.setActive(false);%0A    %7D%0A%7D%0A

Add ShardStateServices to core (copy exists in blueflood-udp, will be removed in next commit)