/*******************************************************************************
 * 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.diffusion.examples.runnable;

import static com.pushtechnology.diffusion.transform.adder.TopicAdderBuilders.binaryTopicAdderBuilder;
import static com.pushtechnology.diffusion.transform.transformer.Transformers.byteArrayToBinary;
import static com.pushtechnology.diffusion.transform.updater.UpdaterBuilders.updaterBuilder;
import static java.util.concurrent.TimeUnit.SECONDS;

import java.nio.ByteBuffer;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;

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

import com.pushtechnology.diffusion.client.callbacks.ErrorReason;
import com.pushtechnology.diffusion.client.features.control.topics.TopicControl;
import com.pushtechnology.diffusion.client.features.control.topics.TopicUpdateControl;
import com.pushtechnology.diffusion.client.session.Session;
import com.pushtechnology.diffusion.datatype.binary.Binary;
import com.pushtechnology.diffusion.transform.adder.SafeTopicAdder;
import com.pushtechnology.diffusion.transform.transformer.SafeTransformer;
import com.pushtechnology.diffusion.transform.updater.SafeTransformedUpdater;


/**
 * A client that creates and updates Binary topics.
 *
 * @author Push Technology Limited
 */
public final class ProducingBinary extends AbstractClient {
    private static final Logger LOG = LoggerFactory.getLogger(ProducingBinary.class);
    private static final SafeTransformer<RandomData, byte[]> SERIALISER = new SafeTransformer<RandomData, byte[]>() {
        @Override
        public byte[] transform(RandomData value) {
            final ByteBuffer buffer = ByteBuffer.allocate(16);
            buffer.putInt(value.getId());
            buffer.putLong(value.getTimestamp());
            buffer.putInt(value.getRandomInt());
            return buffer.array();
        }
    };
    private static final ScheduledExecutorService EXECUTOR = Executors.newSingleThreadScheduledExecutor();

    private volatile Future<?> updateTask;

    /**
     * Constructor.
     * @param url The URL to connect to
     * @param principal The principal to connect as
     */
    public ProducingBinary(String url, String principal) {
        super(url, principal);
    }

    @Override
    public void onConnected(Session session) {
        final SafeTopicAdder<RandomData> adder = binaryTopicAdderBuilder()
            .transform(byteArrayToBinary())
            .transform(SERIALISER)
            .bind(session)
            .create();

        adder.add("binary/random", RandomData.next(), new TopicControl.AddCallback.Default());

        final TopicUpdateControl.Updater updater = session
            .feature(TopicUpdateControl.class)
            .updater();

        // Create a one-way transforming value updater that cannot be used to lookup cached values
        final SafeTransformedUpdater<Binary, RandomData> valueUpdater = updaterBuilder(Binary.class)
            .transform(byteArrayToBinary())
            .transform(SERIALISER)
            .create(updater);

        updateTask = EXECUTOR.scheduleAtFixedRate(
            () -> {
                valueUpdater.update(
                    "binary/random",
                    RandomData.next(),
                    new TopicUpdateControl.Updater.UpdateCallback.Default());
            },
            0L,
            1L,
            SECONDS);
    }

    @Override
    public void onDisconnected() {
        updateTask.cancel(false);
    }

    @Override
    public void onError(ErrorReason errorReason) {
        LOG.error("Failed to start client: {}", errorReason);
    }

    /**
     * Entry point for the example.
     * @param args The command line arguments
     * @throws InterruptedException If the main thread was interrupted
     */
    // CHECKSTYLE.OFF: UncommentedMain
    public static void main(String[] args) throws InterruptedException {
        final ProducingBinary client =
            new ProducingBinary("ws://diffusion.example.com:80", "auth");
        client.start("auth_secret");
        client.waitForStopped();
    }
    // CHECKSTYLE.ON: UncommentedMain
}

@@ -1864,24 +1864,96 @@
 ransformer;%0A
+import com.pushtechnology.diffusion.transform.transformer.Transformers;%0A
 import com.p
@@ -2018,16 +2018,16 @@
 pdater;%0A
-
 %0A%0A/**%0A *
@@ -2317,22 +2317,22 @@
 omData, 
-byte%5B%5D
+Binary
 %3E SERIAL
@@ -2338,16 +2338,83 @@
 LISER = 
+Transformers%0A        .builder(RandomData.class)%0A        .transform(
 new Safe
@@ -2437,22 +2437,26 @@
 omData, 
-byte%5B%5D
+ByteBuffer
 %3E() %7B%0A  
@@ -2457,24 +2457,28 @@
 ) %7B%0A        
+    
 @Override%0A  
@@ -2487,21 +2487,29 @@
     
+    
 public 
-byte%5B%5D
+ByteBuffer
  tra
@@ -2531,24 +2531,28 @@
 ta value) %7B%0A
+    
             
@@ -2606,32 +2606,36 @@
 6);%0A            
+    
 buffer.putInt(va
@@ -2652,32 +2652,36 @@
 ));%0A            
+    
 buffer.putLong(v
@@ -2706,32 +2706,36 @@
 ));%0A            
+    
 buffer.putInt(va
@@ -2767,16 +2767,20 @@
         
+    
 return b
@@ -2788,17 +2788,72 @@
 ffer
-.
+;%0A            %7D%0A        %7D)%0A        .transform(ByteBuffer::
 array
-();
+)
 %0A   
@@ -2861,15 +2861,56 @@
     
-%7D%0A    %7D
+.transform(byteArrayToBinary())%0A        .build()
 ;%0A  
@@ -3424,52 +3424,8 @@
 r()%0A
-            .transform(byteArrayToBinary())%0A
     
@@ -3832,16 +3832,16 @@
  values%0A
+
         
@@ -3937,52 +3937,8 @@
 ss)%0A
-            .transform(byteArrayToBinary())%0A
     

Update example to show usage of transformer builders.