@@ -0,0 +1,2955 @@
+/*%0A * Copyright 2013 the original author or authors.%0A *%0A * Licensed under the Apache License, Version 2.0 (the %22License%22);%0A * you may not use this file except in compliance with the License.%0A * You may obtain a copy of the License at%0A *%0A * http://www.apache.org/licenses/LICENSE-2.0%0A *%0A * Unless required by applicable law or agreed to in writing, software%0A * distributed under the License is distributed on an %22AS IS%22 BASIS,%0A * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.%0A * See the License for the specific language governing permissions and%0A * limitations under the License.%0A */%0Apackage net.kuujo.vertigo.feeder;%0A%0Aimport net.kuujo.vertigo.Vertigo;%0Aimport net.kuujo.vertigo.context.InstanceContext;%0A%0Aimport org.vertx.java.core.AsyncResult;%0Aimport org.vertx.java.core.Future;%0Aimport org.vertx.java.core.Handler;%0Aimport org.vertx.java.core.json.JsonObject;%0Aimport org.vertx.java.platform.Verticle;%0A%0A/**%0A * A Java feeder verticle.%0A *%0A * @author Jordan Halterman%0A */%0Apublic abstract class FeederVerticle extends Verticle %7B%0A  private Vertigo vertigo;%0A  protected PollingFeeder feeder;%0A  protected InstanceContext context;%0A%0A  private Handler%3CString%3E ackHandler = new Handler%3CString%3E() %7B%0A    @Override%0A    public void handle(String messageId) %7B%0A      ack(messageId);%0A    %7D%0A  %7D;%0A%0A  private Handler%3CString%3E failHandler = new Handler%3CString%3E() %7B%0A    @Override%0A    public void handle(String messageId) %7B%0A      fail(messageId);%0A    %7D%0A  %7D;%0A%0A  @Override%0A  public void start(final Future%3CVoid%3E future) %7B%0A    vertigo = new Vertigo(this);%0A    feeder = vertigo.createPollingFeeder().ackHandler(ackHandler).failHandler(failHandler);%0A    context = feeder.getContext();%0A    feeder.start(new Handler%3CAsyncResult%3CPollingFeeder%3E%3E() %7B%0A      @Override%0A      public void handle(AsyncResult%3CPollingFeeder%3E result) %7B%0A        if (result.failed()) %7B%0A          future.setFailure(result.cause());%0A        %7D%0A        else %7B%0A          FeederVerticle.super.start(future);%0A        %7D%0A      %7D%0A    %7D);%0A  %7D%0A%0A  /**%0A   * Called when the feeder is prepared to emit the next message.%0A   */%0A  protected abstract void nextMessage();%0A%0A  /**%0A   * Emits a message from the feeder.%0A   *%0A   * @param data%0A   *   The message body.%0A   * @return%0A   *   The unique message identifier.%0A   */%0A  protected String emit(JsonObject data) %7B%0A    return feeder.emit(data);%0A  %7D%0A%0A  /**%0A   * Emits a message from the feeder.%0A   *%0A   * @param data%0A   *   The message body.%0A   * @param tag%0A   *   A tag to apply to the message.%0A   * @return%0A   *   The unique message identifier.%0A   */%0A  protected String emit(JsonObject data, String tag) %7B%0A    return feeder.emit(data, tag);%0A  %7D%0A%0A  /**%0A   * Called when a message has been acked.%0A   *%0A   * @param id%0A   *   The acked message identifier.%0A   */%0A  protected abstract void ack(String id);%0A%0A  /**%0A   * Called when a message has been failed.%0A   *%0A   * @param id%0A   *   The failed message identifier.%0A   */%0A  protected abstract void fail(String id);%0A%0A%7D%0A

Add java feeder verticle implementation.