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

import com.addthis.bundle.core.Bundle;
import com.addthis.bundle.core.BundleField;
import com.addthis.bundle.core.BundleFormat;
import com.addthis.bundle.value.ValueFactory;
import com.addthis.bundle.value.ValueMap;
import com.addthis.bundle.value.ValueMapEntry;
import com.addthis.bundle.value.ValueObject;
import com.addthis.codec.annotations.FieldConfig;

/**
 * This StreamBuilder splits a ValueMap amongst multiple bundles.
 * Each bundle will have the same fields as the original with the exception
 * of the field specified to use as the map, which won't be duplicated
 * and will have one entry in each bundle with names as specified by keyField and valueField.
 */
public class StreamMapSplitBuilder extends StreamBuilder {
    /** The map field to split into multiple bundles */
    @FieldConfig(required = true) private String field;
    /** The field name in which to place each key in the new bundle */
    @FieldConfig(required = true) private String keyField;
    /** The field name in which to place each value in the new bundle */
    @FieldConfig(required = true) private String valueField;

    @Override
    public void init() {}

    @Override
    public void process(Bundle bundle, StreamEmitter emitter) {
        ValueObject obj = bundle.getValue(bundle.getFormat().getField(field));
        if (obj != null && obj.getObjectType() == ValueObject.TYPE.MAP) {
            ValueMap map = obj.asMap();
            for (ValueMapEntry entry : map) {
                Bundle newBundle = bundle.createBundle();
                BundleFormat format = newBundle.getFormat();
                for (BundleField bundleField : format) {
                    // do not add map field to the new bundle, but do add all others
                    if (!bundleField.getName().equals(obj)) {
                        newBundle.setValue(bundleField, bundle.getValue(bundleField));
                    }
                }
                newBundle.setValue(format.getField(keyField), ValueFactory.create(entry.getKey()));
                newBundle.setValue(format.getField(valueField), entry.getValue());
                emitter.emit(newBundle);
            }
        }
    }
}
@@ -664,21 +664,17 @@
 e.Bundle
-Field
+s
 ;%0Aimport
@@ -697,25 +697,22 @@
 dle.
-core.BundleFormat
+util.AutoField
 ;%0Aim
@@ -889,16 +889,17 @@
 Object;%0A
+%0A
 import c
@@ -905,45 +905,49 @@
 com.
-addthis.codec.annotations.FieldConfig
+fasterxml.jackson.annotation.JsonProperty
 ;%0A%0A/
@@ -1376,35 +1376,36 @@
 les */%0A    @
-FieldConfig
+JsonProperty
 (required = 
@@ -1418,22 +1418,25 @@
 private 
-String
+AutoField
  field;%0A
@@ -1507,35 +1507,36 @@
 dle */%0A    @
-FieldConfig
+JsonProperty
 (required = 
@@ -1549,22 +1549,25 @@
 private 
-String
+AutoField
  keyFiel
@@ -1651,19 +1651,20 @@
    @
-FieldConfig
+JsonProperty
 (req
@@ -1689,14 +1689,17 @@
 ate 
-String
+AutoField
  val
@@ -1853,22 +1853,21 @@
 t obj = 
-bundle
+field
 .getValu
@@ -1878,36 +1878,43 @@
 ndle
-.getFormat().getField(field)
+);%0A        field.removeValue(bundle
 );%0A 
@@ -1923,16 +1923,17 @@
      if 
+(
 (obj != 
@@ -1940,12 +1940,14 @@
 null
+)
  && 
+(
 obj.
@@ -1986,16 +1986,17 @@
 YPE.MAP)
+)
  %7B%0A     
@@ -2115,483 +2115,86 @@
 e = 
-b
+B
 undle
-.createBundle();%0A                BundleFormat format = newBundle.getFormat();%0A                for (BundleField bundleField : format) %7B%0A                    // do not add map field to the new bundle, but do add all others%0A                    if (!bundleField.getName().equals(obj)) %7B%0A                        newBundle.setValue(bundleField, bundle.getValue(bundleField));%0A                    %7D%0A                %7D%0A                newBundle.setValue(format.getField(keyField)
+s.shallowCopyBundle(bundle);%0A                keyField.setValue(newBundle
 , Va
@@ -2249,54 +2249,37 @@
     
-newBundle.setValue(format.getField(valueField)
+valueField.setValue(newBundle
 , en

clean up map split builder