Merge branch 'move-firehose' of github.com:metamx/druid into move-fir…

…ehose
guod08 · Apr 25, 2014 · 171d20d · 171d20d
2 parents eef034c + cd1c1d2
commit 171d20d
Show file tree

Hide file tree

Showing 4 changed files with 24 additions and 3 deletions.
diff --git a/.../worker/executor/ChatHandlerResource.java → ...ealtime/firehose/ChatHandlerResource.java b/.../worker/executor/ChatHandlerResource.java → ...ealtime/firehose/ChatHandlerResource.java
@@ -17,7 +17,7 @@
  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  */
 
-package io.druid.indexing.worker.executor;
+package io.druid.segment.realtime.firehose;
 
 import com.google.common.base.Optional;
 import com.google.inject.Inject;

diff --git a/server/src/main/java/io/druid/segment/realtime/firehose/EventReceiverFirehoseFactory.java b/server/src/main/java/io/druid/segment/realtime/firehose/EventReceiverFirehoseFactory.java
@@ -32,6 +32,7 @@
 import io.druid.data.input.Firehose;
 import io.druid.data.input.FirehoseFactory;
 import io.druid.data.input.InputRow;
+import io.druid.data.input.Rows;
 import io.druid.data.input.impl.MapInputRowParser;
 import io.druid.segment.realtime.firehose.ChatHandler;
 import io.druid.segment.realtime.firehose.ChatHandlerProvider;
@@ -139,7 +140,8 @@ public Response addAll(Collection<Map<String, Object>> events)
       final List<InputRow> rows = Lists.newArrayList();
       for (final Map<String, Object> event : events) {
         // Might throw an exception. We'd like that to happen now, instead of while adding to the row buffer.
-        rows.add(parser.parse(event));
+        InputRow row = parser.parse(event);
+        rows.add(Rows.toCaseInsensitiveInputRow(row,row.getDimensions()));
       }
 
       try {

diff --git a/services/src/main/java/io/druid/cli/CliPeon.java b/services/src/main/java/io/druid/cli/CliPeon.java
@@ -56,7 +56,7 @@
 import io.druid.indexing.overlord.TaskRunner;
 import io.druid.indexing.overlord.TaskStorage;
 import io.druid.indexing.overlord.ThreadPoolTaskRunner;
-import io.druid.indexing.worker.executor.ChatHandlerResource;
+import io.druid.segment.realtime.firehose.ChatHandlerResource;
 import io.druid.indexing.worker.executor.ExecutorLifecycle;
 import io.druid.indexing.worker.executor.ExecutorLifecycleConfig;
 import io.druid.query.QuerySegmentWalker;

diff --git a/services/src/main/java/io/druid/guice/RealtimeModule.java b/services/src/main/java/io/druid/guice/RealtimeModule.java
@@ -31,6 +31,10 @@
 import io.druid.segment.realtime.NoopSegmentPublisher;
 import io.druid.segment.realtime.RealtimeManager;
 import io.druid.segment.realtime.SegmentPublisher;
+import io.druid.segment.realtime.firehose.ChatHandlerProvider;
+import io.druid.segment.realtime.firehose.ChatHandlerResource;
+import io.druid.segment.realtime.firehose.NoopChatHandlerProvider;
+import io.druid.segment.realtime.firehose.ServiceAnnouncingChatHandlerProvider;
 import io.druid.server.QueryResource;
 import io.druid.server.initialization.JettyServerInitializer;
 import org.eclipse.jetty.server.Server;
@@ -57,6 +61,20 @@ public void configure(Binder binder)
     publisherBinder.addBinding("noop").to(NoopSegmentPublisher.class);
     binder.bind(DbSegmentPublisher.class).in(LazySingleton.class);
 
+    PolyBind.createChoice(
+        binder,
+        "druid.realtime.chathandler.type",
+        Key.get(ChatHandlerProvider.class),
+        Key.get(NoopChatHandlerProvider.class)
+    );
+    final MapBinder<String, ChatHandlerProvider> handlerProviderBinder = PolyBind.optionBinder(
+        binder, Key.get(ChatHandlerProvider.class)
+    );
+    handlerProviderBinder.addBinding("announce")
+                         .to(ServiceAnnouncingChatHandlerProvider.class).in(LazySingleton.class);
+    handlerProviderBinder.addBinding("noop")
+                         .to(NoopChatHandlerProvider.class).in(LazySingleton.class);
+
     JsonConfigProvider.bind(binder, "druid.realtime", RealtimeManagerConfig.class);
     binder.bind(new TypeLiteral<List<FireDepartment>>(){})
           .toProvider(FireDepartmentsProvider.class)
@@ -66,6 +84,7 @@ public void configure(Binder binder)
     binder.bind(NodeTypeConfig.class).toInstance(new NodeTypeConfig("realtime"));
     binder.bind(JettyServerInitializer.class).to(QueryJettyServerInitializer.class).in(LazySingleton.class);
     Jerseys.addResource(binder, QueryResource.class);
+    Jerseys.addResource(binder, ChatHandlerResource.class);
     LifecycleModule.register(binder, QueryResource.class);
 
     LifecycleModule.register(binder, Server.class);