mirror of
https://github.com/apache/druid.git
synced 2025-02-17 15:35:56 +00:00
Chat handlers still don't work, but, they're closer maybe.
This commit is contained in:
parent
59f2d0711d
commit
384dcda7e4
@ -20,7 +20,7 @@
|
|||||||
package io.druid.indexing.common.index;
|
package io.druid.indexing.common.index;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Objects that can be registered with a {@link EventReceivingChatHandlerProvider} and provide http endpoints for indexing-related
|
* Objects that can be registered with a {@link ServiceAnnouncingChatHandlerProvider} and provide http endpoints for indexing-related
|
||||||
* objects. This interface is empty because it only exists to signal intent. The actual http endpoints are provided
|
* objects. This interface is empty because it only exists to signal intent. The actual http endpoints are provided
|
||||||
* through JAX-RS annotations on the {@link ChatHandler} objects.
|
* through JAX-RS annotations on the {@link ChatHandler} objects.
|
||||||
*/
|
*/
|
||||||
|
@ -25,8 +25,6 @@ import com.google.common.base.Optional;
|
|||||||
*/
|
*/
|
||||||
public interface ChatHandlerProvider
|
public interface ChatHandlerProvider
|
||||||
{
|
{
|
||||||
public String getType();
|
|
||||||
|
|
||||||
public void register(final String key, ChatHandler handler);
|
public void register(final String key, ChatHandler handler);
|
||||||
|
|
||||||
public void unregister(final String key);
|
public void unregister(final String key);
|
||||||
|
@ -49,7 +49,7 @@ import java.util.concurrent.TimeUnit;
|
|||||||
|
|
||||||
/**
|
/**
|
||||||
* Builds firehoses that accept events through the {@link EventReceiver} interface. Can also register these
|
* Builds firehoses that accept events through the {@link EventReceiver} interface. Can also register these
|
||||||
* firehoses with an {@link EventReceivingChatHandlerProvider}.
|
* firehoses with an {@link ServiceAnnouncingChatHandlerProvider}.
|
||||||
*/
|
*/
|
||||||
@JsonTypeName("receiver")
|
@JsonTypeName("receiver")
|
||||||
public class EventReceiverFirehoseFactory implements FirehoseFactory
|
public class EventReceiverFirehoseFactory implements FirehoseFactory
|
||||||
@ -84,8 +84,9 @@ public class EventReceiverFirehoseFactory implements FirehoseFactory
|
|||||||
final EventReceiverFirehose firehose = new EventReceiverFirehose();
|
final EventReceiverFirehose firehose = new EventReceiverFirehose();
|
||||||
|
|
||||||
if (chatHandlerProvider.isPresent()) {
|
if (chatHandlerProvider.isPresent()) {
|
||||||
log.info("Found chathandler with type[%s]", chatHandlerProvider.get().getType());
|
log.info("Found chathandler of class[%s]", chatHandlerProvider.get().getClass().getName());
|
||||||
chatHandlerProvider.get().register(serviceName, firehose);
|
chatHandlerProvider.get().register(serviceName, firehose);
|
||||||
|
chatHandlerProvider.get().register(serviceName.replaceAll(".*:", ""), firehose); // rolf
|
||||||
} else {
|
} else {
|
||||||
log.info("No chathandler detected");
|
log.info("No chathandler detected");
|
||||||
}
|
}
|
||||||
|
@ -25,12 +25,6 @@ import com.google.common.base.Optional;
|
|||||||
*/
|
*/
|
||||||
public class NoopChatHandlerProvider implements ChatHandlerProvider
|
public class NoopChatHandlerProvider implements ChatHandlerProvider
|
||||||
{
|
{
|
||||||
@Override
|
|
||||||
public String getType()
|
|
||||||
{
|
|
||||||
return "noop";
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void register(String key, ChatHandler handler)
|
public void register(String key, ChatHandler handler)
|
||||||
{
|
{
|
||||||
|
@ -35,16 +35,16 @@ import java.util.concurrent.ConcurrentMap;
|
|||||||
* allows anyone with a reference to this object to obtain a particular {@link ChatHandler}. An embedded
|
* allows anyone with a reference to this object to obtain a particular {@link ChatHandler}. An embedded
|
||||||
* {@link ServiceAnnouncer} will be used to advertise handlers on this host.
|
* {@link ServiceAnnouncer} will be used to advertise handlers on this host.
|
||||||
*/
|
*/
|
||||||
public class EventReceivingChatHandlerProvider implements ChatHandlerProvider
|
public class ServiceAnnouncingChatHandlerProvider implements ChatHandlerProvider
|
||||||
{
|
{
|
||||||
private static final Logger log = new Logger(EventReceivingChatHandlerProvider.class);
|
private static final Logger log = new Logger(ServiceAnnouncingChatHandlerProvider.class);
|
||||||
|
|
||||||
private final DruidNode node;
|
private final DruidNode node;
|
||||||
private final ServiceAnnouncer serviceAnnouncer;
|
private final ServiceAnnouncer serviceAnnouncer;
|
||||||
private final ConcurrentMap<String, ChatHandler> handlers;
|
private final ConcurrentMap<String, ChatHandler> handlers;
|
||||||
|
|
||||||
@Inject
|
@Inject
|
||||||
public EventReceivingChatHandlerProvider(
|
public ServiceAnnouncingChatHandlerProvider(
|
||||||
@Self DruidNode node,
|
@Self DruidNode node,
|
||||||
ServiceAnnouncer serviceAnnouncer
|
ServiceAnnouncer serviceAnnouncer
|
||||||
)
|
)
|
@ -45,8 +45,9 @@ import io.druid.indexing.common.actions.RemoteTaskActionClientFactory;
|
|||||||
import io.druid.indexing.common.actions.TaskActionClientFactory;
|
import io.druid.indexing.common.actions.TaskActionClientFactory;
|
||||||
import io.druid.indexing.common.config.TaskConfig;
|
import io.druid.indexing.common.config.TaskConfig;
|
||||||
import io.druid.indexing.common.index.ChatHandlerProvider;
|
import io.druid.indexing.common.index.ChatHandlerProvider;
|
||||||
import io.druid.indexing.common.index.EventReceivingChatHandlerProvider;
|
import io.druid.indexing.common.index.ServiceAnnouncingChatHandlerProvider;
|
||||||
import io.druid.indexing.common.index.NoopChatHandlerProvider;
|
import io.druid.indexing.common.index.NoopChatHandlerProvider;
|
||||||
|
import io.druid.indexing.common.index.ServiceAnnouncingChatHandlerProvider;
|
||||||
import io.druid.indexing.coordinator.TaskRunner;
|
import io.druid.indexing.coordinator.TaskRunner;
|
||||||
import io.druid.indexing.coordinator.ThreadPoolTaskRunner;
|
import io.druid.indexing.coordinator.ThreadPoolTaskRunner;
|
||||||
import io.druid.indexing.worker.executor.ChatHandlerResource;
|
import io.druid.indexing.worker.executor.ChatHandlerResource;
|
||||||
@ -104,8 +105,8 @@ public class CliPeon extends GuiceRunnable
|
|||||||
final MapBinder<String, ChatHandlerProvider> handlerProviderBinder = PolyBind.optionBinder(
|
final MapBinder<String, ChatHandlerProvider> handlerProviderBinder = PolyBind.optionBinder(
|
||||||
binder, Key.get(ChatHandlerProvider.class)
|
binder, Key.get(ChatHandlerProvider.class)
|
||||||
);
|
);
|
||||||
handlerProviderBinder.addBinding("receiver")
|
handlerProviderBinder.addBinding("announce")
|
||||||
.to(EventReceivingChatHandlerProvider.class).in(LazySingleton.class);
|
.to(ServiceAnnouncingChatHandlerProvider.class).in(LazySingleton.class);
|
||||||
handlerProviderBinder.addBinding("noop")
|
handlerProviderBinder.addBinding("noop")
|
||||||
.to(NoopChatHandlerProvider.class).in(LazySingleton.class);
|
.to(NoopChatHandlerProvider.class).in(LazySingleton.class);
|
||||||
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user