/*
 * 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.addthis.hydra.task.output;

import java.util.List;

import com.addthis.bundle.channel.DataChannelError;
import com.addthis.bundle.core.Bundle;
import com.addthis.bundle.core.Bundles;
import com.addthis.codec.Codec;
import com.addthis.hydra.task.run.TaskRunConfig;

import org.slf4j.Logger;


import org.slf4j.LoggerFactory;
/**
 * This output sink <span class="hydra-summary">writes to multiple output sinks</span>.
 * <p/>
 * <p>Each bundle will be emitted to the first sink, and then the second sink, etc.
 * The {@link #immutableCopy immutableCopy} parameter can be specified if one
 * of the children sinks will modify a bundle, and you do not wish those modifications
 * to appear in the remaining children sinks.
 * <p/>
 * <p>Example:</p>
 * <pre>output : {
 *   type : "chain",
 *   outputs : [
 *     {
 *       type : "file",
 *       path : [ "{{DATE_YMD}}", "/","{{PATH_TYPE}}", "/","{{SHARD}}"],
 *       ...
 *     },
 *     {
 *       type : "tree",
 *       stats : true,
 *       root : {path : "ROOT"},
 *       ...
 *     }
 *   ]
 * }</pre>
 *
 * @user-reference
 * @hydra-name chain
 */
public class TaskDataOutputChain extends TaskDataOutput {

    private static final Logger log = LoggerFactory.getLogger(TaskDataOutputChain.class);

    /**
     * Sequence of output sinks. Each bundle is emitted to the first sink,
     * then the second sink, etc.
     */
    @Codec.Set(codable = true, required = true)
    private TaskDataOutput outputs[];

    /**
     * If true then create a deep copy of a bundle when it is passed to a child
     * output sink. This may be useful when the child sink modifies
     * the bundle. Default value is false.
     */
    @Codec.Set(codable = true)
    private boolean immutableCopy = false;

    private TaskDataOutput currentOutput = null;

    @Override
    protected void open(TaskRunConfig config) {
        log.warn("[init] beginning init chain");
        for (int i = 0; i < outputs.length; i++) {
            outputs[i].open(config);
        }
        currentOutput = outputs[0];
        log.warn("[init] all outputs initialized");
    }


    public void send(Bundle row) throws DataChannelError {
        if (immutableCopy) {
            for (TaskDataOutput output : outputs) {
                currentOutput = output;
                output.send(Bundles.deepCopyBundle(row, output.createBundle()));
            }
        } else {
            for (TaskDataOutput output : outputs) {
                currentOutput = output;
                output.send(Bundles.shallowCopyBundle(row, output.createBundle()));
            }
        }
    }


    public void send(List<Bundle> bundles) {
        if (bundles != null && !bundles.isEmpty()) {
            for (Bundle bundle : bundles) {
                send(bundle);
            }
        }
    }


    public void sendComplete() {
        log.warn("[sendComplete] forwarding completion signal to all outputs");
        for (TaskDataOutput output : outputs) {
            currentOutput = output;
            output.sendComplete();
        }
        log.warn("[sendComplete] forwarding complete");
    }

    public void sourceError(DataChannelError er) {
        log.warn("[sourceError] forwarding to all outputs" + er);
        for (TaskDataOutput output : outputs) {
            currentOutput = output;
            output.sourceError(er);
        }
        log.warn("[sourceError] forwarding complete");
    }

    @Override
    public Bundle createBundle() {
        return currentOutput.createBundle();
    }
}

@@ -1712,20 +1712,16 @@
 extends 
-Task
 DataOutp
@@ -1722,16 +1722,24 @@
 taOutput
+TypeList
  %7B%0A%0A    
@@ -2323,58 +2323,8 @@
 e;%0A%0A
-    private TaskDataOutput currentOutput = null;%0A%0A
     
@@ -2532,44 +2532,8 @@
   %7D%0A
-        currentOutput = outputs%5B0%5D;%0A
     
@@ -2579,33 +2579,32 @@
 lized%22);%0A    %7D%0A%0A
-%0A
     public void 
@@ -2731,48 +2731,8 @@
 ) %7B%0A
-                currentOutput = output;%0A
     
@@ -2895,48 +2895,8 @@
 ) %7B%0A
-                currentOutput = output;%0A
     
@@ -2998,33 +2998,32 @@
        %7D%0A    %7D%0A%0A
-%0A
     public void 
@@ -3209,17 +3209,16 @@
 %0A    %7D%0A%0A
-%0A
     publ
@@ -3374,44 +3374,8 @@
 ) %7B%0A
-            currentOutput = output;%0A
     
@@ -3647,44 +3647,8 @@
 ) %7B%0A
-            currentOutput = output;%0A
     
@@ -3681,32 +3681,32 @@
 (er);%0A        %7D%0A
+
         log.warn
@@ -3753,108 +3753,7 @@
    %7D
-%0A%0A    @Override%0A    public Bundle createBundle() %7B%0A        return currentOutput.createBundle();%0A    %7D
 %0A%7D%0A

remove currentOutput silliness