/*******************************************************************************
 * Copyright (C) 2016 Push Technology Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *******************************************************************************/

package com.pushtechnology.adapters.rest.adapter;

import java.util.ArrayList;
import java.util.List;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.http.concurrent.FutureCallback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.pushtechnology.adapters.rest.model.latest.Model;
import com.pushtechnology.adapters.rest.model.latest.ServiceConfig;
import com.pushtechnology.adapters.rest.polling.EndpointClient;
import com.pushtechnology.adapters.rest.polling.EndpointResponse;
import com.pushtechnology.adapters.rest.polling.ServiceSession;
import com.pushtechnology.adapters.rest.polling.ServiceSessionFactory;
import com.pushtechnology.adapters.rest.publication.PublishingClient;
import com.pushtechnology.adapters.rest.topic.management.TopicManagementClient;
import com.pushtechnology.diffusion.client.features.control.topics.TopicAddFailReason;
import com.pushtechnology.diffusion.client.features.control.topics.TopicControl;

/**
 * Implementation for {@link ServiceSessionGroup}.
 *
 * @author Push Technology Limited
 */
public final class ServiceSessionGroupImpl implements ServiceSessionGroup {
    private static final Logger LOG = LoggerFactory.getLogger(ServiceSessionGroupImpl.class);
    private final Model model;
    private final TopicManagementClient topicManagementClient;
    private final EndpointClient endpointClient;
    private final PublishingClient publishingClient;
    private final List<ServiceSession> serviceSessions;
    private final ServiceSessionFactory serviceSessionFactory;
    private final ServiceListener serviceListener;

    /**
     * Constructor.
     */
    public ServiceSessionGroupImpl(
            Model model,
            TopicManagementClient topicManagementClient,
            EndpointClient endpointClient,
            PublishingClient publishingClient,
            ServiceSessionFactory serviceSessionFactory,
            ServiceListener serviceListener) {
        this.model = model;
        this.topicManagementClient = topicManagementClient;
        this.endpointClient = endpointClient;
        this.publishingClient = publishingClient;
        this.serviceSessionFactory = serviceSessionFactory;
        this.serviceListener = serviceListener;
        this.serviceSessions = new ArrayList<>();
    }

    @PostConstruct
    @Override
    public synchronized void start() {
        LOG.debug("Opening service session group");
        for (final ServiceConfig service : model.getServices()) {
            final ServiceSession serviceSession = serviceSessionFactory.create(service);
            topicManagementClient.addService(service);
            publishingClient
                .addService(service)
                .onStandby(() -> {
                    LOG.info("Service {} on standby", service);
                    serviceListener.onStandby(service);
                })
                .onActive((updater) -> {
                    LOG.info("Service {} active", service);
                    serviceListener.onActive(service);
                    service.getEndpoints().forEach(endpoint -> {
                        endpointClient.request(service, endpoint, new FutureCallback<EndpointResponse>() {
                            @Override
                            public void completed(EndpointResponse result) {
                                topicManagementClient.addEndpoint(service, endpoint, new TopicControl.AddCallback() {
                                    @Override
                                    public void onTopicAdded(String topicPath) {
                                        LOG.info("Endpoint {} exists, adding endpoint to service session", endpoint);
                                        serviceSession.addEndpoint(endpoint);
                                    }

                                    @Override
                                    public void onTopicAddFailed(String topicPath, TopicAddFailReason reason) {
                                        LOG.warn("Topic creation failed for {} because {}", endpoint, reason);
                                    }

                                    @Override
                                    public void onDiscard() {
                                        LOG.warn("Topic creation for {} might not have succeeded", endpoint);
                                    }
                                });
                            }

                            @Override
                            public void failed(Exception ex) {
                                LOG.warn("Initial request to {} failed", endpoint, ex);
                            }

                            @Override
                            public void cancelled() {
                                LOG.warn("Initial request to {} cancelled", endpoint);
                            }
                        });
                    });
                    serviceSession.start();
                })
                .onClose(() -> {
                    LOG.info("Service {} closed", service);
                    serviceListener.onRemove(service);
                });
            serviceSessions.add(serviceSession);
        }
        LOG.debug("Opened service session group");
    }

    @PreDestroy
    @Override
    public synchronized void close() {
        LOG.debug("Closing service session group");
        serviceSessions.forEach(ServiceSession::stop);
        model.getServices().forEach(publishingClient::removeService);
        LOG.debug("Closed service session group");
    }
}

@@ -1034,24 +1034,93 @@
 erFactory;%0A%0A
+import com.pushtechnology.adapters.rest.model.latest.EndpointConfig;%0A
 import com.p
@@ -4163,24 +4163,181 @@
 e result) %7B%0A
+                                if (!validateContentType(result, endpoint)) %7B%0A                                    return;%0A                                %7D%0A%0A
             
@@ -6260,24 +6260,935 @@
 p%22);%0A    %7D%0A%0A
+    private boolean validateContentType(EndpointResponse result, EndpointConfig endpoint) %7B%0A        final String contentType = result.getHeader(%22content-type%22);%0A        if (contentType == null) %7B%0A            // Assume correct when no content type provided%0A            return true;%0A        %7D%0A%0A        if (%22binary%22.equals(endpoint.getProduces())) %7B%0A            // Everything is binary%0A            return true;%0A        %7D%0A%0A        if (contentType.startsWith(endpoint.getProduces())) %7B%0A            // Content type looks about right%0A            return true;%0A        %7D%0A%0A        if (contentType.startsWith(%22application/json%22) %7C%7C contentType.startsWith(%22text/json%22)) %7B%0A            // JSON content can be treated as binary, string, or JSON%0A            return true;%0A        %7D%0A%0A        LOG.warn(%22The content type of the response %7B%7D is not suitable for the endpoint %7B%7D%22, contentType, endpoint);%0A        return false;%0A    %7D%0A%0A
     @PreDest

Check the content type of the initial response against the type expected by the endpoint.