/*
 * Copyright © 2015 Cask Data, Inc.
 *
 * 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 co.cask.cdap.templates.etl.batch.sources;

import co.cask.cdap.api.data.schema.Schema;
import co.cask.cdap.api.data.stream.Stream;
import co.cask.cdap.api.data.stream.StreamBatchReadable;
import co.cask.cdap.api.flow.flowlet.StreamEvent;
import co.cask.cdap.templates.etl.api.Property;
import co.cask.cdap.templates.etl.api.StageConfigurer;
import co.cask.cdap.templates.etl.api.batch.BatchSource;
import co.cask.cdap.templates.etl.api.batch.BatchSourceContext;
import co.cask.cdap.templates.etl.common.ETLUtils;
import org.apache.hadoop.io.LongWritable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A {@link BatchSource} for {@link Stream} to use {@link Stream} as Source.
 */
public class StreamBatchSource extends BatchSource<LongWritable, StreamEvent> {

  private static final Logger LOG = LoggerFactory.getLogger(StreamBatchSource.class);

  public void configure(StageConfigurer configurer) {
    configurer.setName(StreamBatchSource.class.getSimpleName());
    configurer.setDescription("Use Stream as Source");
    configurer.addProperty(new Property("streamName", "Name of the stream to use as Source", true));
    configurer.addProperty(new Property("frequency", "Frequency of the schedule", false));
  }

  /**
   * Prepare the Batch Job. Used to configure the Hadoop Job before starting the Batch Job.
   *
   * @param context {@link BatchSourceContext}
   */
  @Override
  public void prepareJob(BatchSourceContext context) {
    long endTime = context.getLogicalStartTime();
    //TODO: Once the method to get the frequency from the schedule is added change it to use that. Then we will not
    // need the frequency as a configuration here (depends on to CDAP-2048)
    long startTime = endTime - ETLUtils.parseFrequency(context.getRuntimeArguments().get("frequency"));

    String streamName = context.getRuntimeArguments().get("streamName");
    LOG.info("Setting input to Stream : {}", streamName);
    Schema schema = Schema.recordOf("streamEvent", Schema.Field.of("body", Schema.of(Schema.Type.STRING)));

    // TODO: This is not clean.
    context.setInput(new StreamBatchReadable(streamName, startTime, endTime,
                                             "co.cask.cdap.data.stream.decoder.IdentityStreamEventDecoder")
                       .toURI().toString());
  }
}

@@ -837,16 +837,74 @@
 mEvent;%0A
+import co.cask.cdap.templates.etl.api.PipelineConfigurer;%0A
 import c
@@ -943,16 +943,16 @@
 operty;%0A
-
 import c
@@ -1119,16 +1119,71 @@
 ontext;%0A
+import co.cask.cdap.templates.etl.api.config.ETLStage;%0A
 import c
@@ -1225,16 +1225,61 @@
 LUtils;%0A
+import com.google.common.base.Preconditions;%0A
 import o
@@ -1621,16 +1621,73 @@
 .class);
+%0A  public static final String STREAM_NAME = %22streamName%22;
 %0A%0A  publ
@@ -2057,161 +2057,353 @@
 %0A%0A  
-/**%0A   * Prepare the Batch Job. Used to configure the Hadoop Job before starting the Batch Job.%0A   *%0A   * @param context %7B@link BatchSourceContext%7D%0A   */
+@Override%0A  public void configurePipeline(ETLStage stageConfig, PipelineConfigurer pipelineConfigurer) %7B%0A    String streamName = stageConfig.getProperties().get(STREAM_NAME);%0A    Preconditions.checkArgument(streamName != null && !streamName.isEmpty(), %22Stream name must be given.%22);%0A    pipelineConfigurer.addStream(new Stream(streamName));%0A  %7D%0A
 %0A  @
@@ -2871,28 +2871,27 @@
 s().get(
-%22streamName%22
+STREAM_NAME
 );%0A    L

Add configurePipeline in StreamBatchSource to create stream