
package com.pushtechnology.adapters;

import static com.pushtechnology.diffusion.client.session.SessionAttributes.Transport.WEBSOCKET;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.pushtechnology.adapters.rest.model.v3.Endpoint;
import com.pushtechnology.adapters.rest.model.v3.Service;
import com.pushtechnology.diffusion.client.Diffusion;
import com.pushtechnology.diffusion.client.features.control.topics.TopicControl;
import com.pushtechnology.diffusion.client.features.control.topics.TopicUpdateControl;
import com.pushtechnology.diffusion.client.features.control.topics.TopicUpdateControl.Updater.UpdateCallback;
import com.pushtechnology.diffusion.client.session.Session;
import com.pushtechnology.diffusion.client.topics.details.TopicType;
import com.pushtechnology.diffusion.datatype.json.JSON;

/**
 * Implements {@link PublishingClient}.
 *
 * @author Push Technology Limited
 */
public final class PublishingClientImpl implements PublishingClient {
    private static final Logger LOG = LoggerFactory.getLogger(PublishingClientImpl.class);
    private final String host;
    private final int port;
    private Session session;

    public PublishingClientImpl(String host, int port) {
        this.host = host;
        this.port = port;
    }

    @Override
    public synchronized void start() {
        // Use the session factory to open a new session
        session = Diffusion
            .sessions()
            .listener(new Listener())
            .serverHost(host)
            .serverPort(port)
            .secureTransport(false)
            .transports(WEBSOCKET)
            .open();
    }

    @Override
    public synchronized void initialise(Service service) {
        final TopicControl topicControl = session.feature(TopicControl.class);

        service.getEndpoints().stream().forEach(endpoint -> {
            topicControl.addTopic(endpoint.getTopic(), TopicType.JSON, new TopicControl.AddCallback.Default());
        });
    }

    @Override
    public synchronized void stop() {
        session.close();
    }

    @Override
    public synchronized void publish(Endpoint endpoint, JSON json) {
        if (!session.getState().isConnected()) {
            return;
        }

        session
            .feature(TopicUpdateControl.class)
            .updater()
            .valueUpdater(JSON.class)
            .update(endpoint.getTopic(), json, new UpdateCallback.Default());
    }

    /**
     * A simple session state listener that logs out state changes.
     */
    private static final class Listener implements Session.Listener {

        @Override
        public void onSessionStateChanged(Session session, Session.State oldState, Session.State newState) {
            LOG.info("{} {} -> {}", session, oldState, newState);
        }
    }
}

@@ -128,16 +128,97 @@
 BSOCKET;
+%0Aimport static com.pushtechnology.diffusion.client.topics.details.TopicType.JSON;
 %0A%0Aimport
@@ -513,32 +513,125 @@
 s.TopicControl;%0A
+import com.pushtechnology.diffusion.client.features.control.topics.TopicControl.AddCallback;%0A
 import com.pusht
@@ -875,77 +875,8 @@
 on;%0A
-import com.pushtechnology.diffusion.client.topics.details.TopicType;%0A
 impo
@@ -1907,16 +1907,29 @@
  service
+%0A            
 .getEndp
@@ -1939,17 +1939,21 @@
 ts()
-.stream()
+%0A            
 .for
@@ -1968,30 +1968,16 @@
 point -%3E
- %7B%0A           
  topicCo
@@ -2016,41 +2016,18 @@
 (), 
-TopicType.JSON, new TopicControl.
+JSON, new 
 AddC
@@ -2044,27 +2044,16 @@
 fault())
-;%0A        %7D
 );%0A    %7D

Clean up collection traversal.