/*******************************************************************************
 * 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.transform.stream;

import static com.pushtechnology.diffusion.transform.transformer.Transformers.identity;

import com.pushtechnology.diffusion.datatype.binary.Binary;
import com.pushtechnology.diffusion.datatype.json.JSON;

/**
 * Factory for creating instances of {@link SafeStreamBuilder}s and {@link StreamBuilder}s.
 *
 * @author Push Technology Limited
 */
public final class StreamBuilders {
    private StreamBuilders() {
    }

    /**
     * Create a {@link StreamBuilder} from a source.
     *
     * @param valueType the type value of the source values
     * @param <V> the type of the source values
     * @return The stream builder
     */
    public static <V> SafeStreamBuilder<V, V> newStreamBuilder(Class<V> valueType) {
        return new SafeStreamBuilderImpl<>(valueType, identity(valueType));
    }

    /**
     * Create a {@link StreamBuilder} from a {@link Binary} source.
     *
     * @return The stream builder
     */
    public static SafeStreamBuilder<Binary, Binary> newBinaryStreamBuilder() {
        return new SafeStreamBuilderImpl<>(Binary.class, identity(Binary.class));
    }

    /**
     * Create a {@link StreamBuilder} from a {@link JSON} source.
     *
     * @return The stream builder
     */
    public static SafeStreamBuilder<JSON, JSON> newJsonStreamBuilder() {
        return new SafeStreamBuilderImpl<>(JSON.class, identity(JSON.class));
    }
}

@@ -1011,16 +1011,80 @@
 on.JSON;
+%0Aimport com.pushtechnology.diffusion.datatype.recordv2.RecordV2;
 %0A%0A/**%0A *
@@ -1889,37 +1889,32 @@
       return new
- Safe
 StreamBuilderImp
@@ -1910,22 +1910,16 @@
 mBuilder
-Impl%3C%3E
 (Binary.
@@ -1927,32 +1927,8 @@
 lass
-, identity(Binary.class)
 );%0A 
@@ -2151,65 +2151,1065 @@
  new
- SafeStreamBuilderImpl%3C%3E(JSON.class, identity(JSON.class)
+StreamBuilder(JSON.class);%0A    %7D%0A%0A    /**%0A     * Create a %7B@link StreamBuilder%7D from a %7B@link String%7D source.%0A     *%0A     * @return The stream builder%0A     */%0A    public static SafeStreamBuilder%3CString, String%3E newStringStreamBuilder() %7B%0A        return newStreamBuilder(String.class);%0A    %7D%0A%0A    /**%0A     * Create a %7B@link StreamBuilder%7D from a %7B@link Long%7D source.%0A     *%0A     * @return The stream builder%0A     */%0A    public static SafeStreamBuilder%3CLong, Long%3E newInt64StreamBuilder() %7B%0A        return newStreamBuilder(Long.class);%0A    %7D%0A%0A    /**%0A     * Create a %7B@link StreamBuilder%7D from a %7B@link Double%7D source.%0A     *%0A     * @return The stream builder%0A     */%0A    public static SafeStreamBuilder%3CDouble, Double%3E newDoubleStreamBuilder() %7B%0A        return newStreamBuilder(Double.class);%0A    %7D%0A%0A    /**%0A     * Create a %7B@link StreamBuilder%7D from a %7B@link RecordV2%7D source.%0A     *%0A     * @return The stream builder%0A     */%0A    public static SafeStreamBuilder%3CRecordV2, RecordV2%3E newRecordV2StreamBuilder() %7B%0A        return newStreamBuilder(RecordV2.class
 );%0A 

Add additional stream builders for new topic types.