package io.cattle.platform.agent.impl;

import io.cattle.platform.async.utils.AsyncUtils;
import io.cattle.platform.eventing.EventCallOptions;
import io.cattle.platform.eventing.EventListener;
import io.cattle.platform.eventing.EventService;
import io.cattle.platform.eventing.model.Event;
import io.cattle.platform.eventing.model.EventVO;
import io.cattle.platform.iaas.event.IaasEvents;
import io.cattle.platform.iaas.event.delegate.DelegateEvent;
import io.cattle.platform.json.JsonMapper;

import java.util.Map;

import com.google.common.util.concurrent.ListenableFuture;

public class WrappedEventService implements EventService {

    long agentId;
    boolean delegate;
    EventService eventService;
    Map<String, Object> instanceData;
    EventService agentRequestPublisher;
    JsonMapper jsonMapper;

    public WrappedEventService(long agentId, boolean delegate, EventService eventService, Map<String, Object> instanceData, JsonMapper jsonMapper) {
        super();
        this.agentId = agentId;
        this.delegate = delegate;
        this.eventService = eventService;
        this.instanceData = instanceData;
        this.jsonMapper = jsonMapper;
    }

    protected Event buildEvent(Event request) {
        Event event = null;
        Object payload = request.getData();
        if (payload instanceof Event) {
            event = new EventVO<>((Event)payload).withReplyTo(((Event)payload).getName() + Event.REPLY_SUFFIX);
        }
        if (delegate) {
            event = new DelegateEvent(instanceData, event);
        }
        return new EventVO<>(event).withName(IaasEvents.appendAgent(event.getName(), agentId));
    }

    @Override
    public boolean publish(Event request) {
        return eventService.publish(buildEvent(request));
    }

    @Override
    public ListenableFuture<Event> call(final Event request, EventCallOptions callOptions) {
        Event unwrappedEvent = buildEvent(request);
        if (unwrappedEvent != null && unwrappedEvent.getName() != null &&
                unwrappedEvent.getName().startsWith(IaasEvents.AGENT_CLOSE)) {
            return eventService.call(request, callOptions);
        }

        return EventCallProgressHelper.call(eventService, unwrappedEvent, Event.class, callOptions, new EventResponseMarshaller() {
            @SuppressWarnings("unchecked")
            @Override
            public <T> T convert(Event resultEvent, Class<T> reply) {
                Object payload = resultEvent;
                if (delegate) {
                    payload = resultEvent.getData();
                }
                return (T) new EventVO<>().withName(request.getReplyTo()).withData(payload);
            }
        });
    }

    /* Boilerplate to implementate interface */
    @Override
    public ListenableFuture<?> subscribe(String eventName, EventListener listener) {
        throw new UnsupportedOperationException();
    }

    @Override
    public void unsubscribe(String eventName, EventListener listener) {
        throw new UnsupportedOperationException();
    }

    @Override
    public void unsubscribe(EventListener listener) {
        throw new UnsupportedOperationException();
    }

    @Override
    public Event callSync(Event event, EventCallOptions callOptions) {
        return AsyncUtils.get(call(event, callOptions));
    }

}
@@ -333,16 +333,121 @@
 ventVO;%0A
+import io.cattle.platform.framework.event.Ping;%0Aimport io.cattle.platform.framework.event.data.PingData;%0A
 import i
@@ -1883,110 +1883,178 @@
     
-@Override%0A    public ListenableFuture%3CEvent%3E call(final Event request, EventCallOptions callOptions) %7B
+protected boolean useAgentConnection(Event unwrappedEvent) %7B%0A        if (unwrappedEvent == null %7C%7C unwrappedEvent.getName() == null) %7B%0A            return false;%0A        %7D
 %0A   
@@ -2058,22 +2058,20 @@
         
-Event 
+if (
 unwrappe
@@ -2080,180 +2080,620 @@
 vent
- = buildEvent(request);%0A        if (unwrappedEvent != null && unwrappedEvent.getName() != null &&%0A                unwrappedEvent.getName().startsWith(IaasEvents.AGENT_CLOSE
+.getName().startsWith(IaasEvents.AGENT_CLOSE)) %7B%0A            return true;%0A        %7D%0A        if (unwrappedEvent.getData() instanceof PingData) %7B%0A            Map%3CString, Boolean%3E options = ((PingData) unwrappedEvent.getData()).getOptions();%0A            if (options != null && Boolean.TRUE.equals(options.get(Ping.RESOURCES))) %7B%0A                return true;%0A            %7D%0A        %7D%0A        return false;%0A    %7D%0A%0A    @Override%0A    public ListenableFuture%3CEvent%3E call(final Event request, EventCallOptions callOptions) %7B%0A        Event unwrappedEvent = buildEvent(request);%0A        if (useAgentConnection(unwrappedEvent
 )) %7B

Send pings for resources through agent connection