/*
 *  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.github.sakserv.storm.scheme;

import backtype.storm.spout.Scheme;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import com.github.sakserv.avro.AvroSchemaUtils;
import mypipe.avro.InsertMutation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class AvroMyPipeTestingScheme implements Scheme {

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

    private static final long serialVersionUID = -2990121166902741545L;
    
    private static List<String> fieldsList = new ArrayList<String>();
    
        // 1 byte - magic for version
        // 1 byte - mutation id
        // n bytes - schemaid (this appears to always be a short with a val of 0)
        // n bytes - payload

        @Override
        public List<Object> deserialize(byte[] bytes) {
            
            //Values values = new Values();
            
           /* String mutationType = AvroSchemaUtils.getMutationType(bytes);
            if(mutationType.equals("InsertMutation")) {
                byte[] payload = AvroSchemaUtils.getAvroPayload(bytes);
                try {
                    InsertMutation insertMutation = AvroSchemaUtils.deserializeInsertMutation(payload);
                    
                    // FirstName
                    fieldsList.add("firstname");
                    values.add(AvroSchemaUtils.getStringValueByKey(insertMutation.getStrings(), "firstname"));
                    
                } catch(IOException e) {
                    e.printStackTrace();
                }

            }*/
            
            //values.add("henry");
            //fieldsList.add("firstname");
            
            //values.add("smith");
            //fieldsList.add("lastname");

            return new Values("henry", "smith");
        }

        @Override
            public Fields getOutputFields() {
            return new Fields("firstname", "lastname");
        }
}

@@ -1549,18 +1549,16 @@
         
-//
 Values v
@@ -2291,34 +2291,32 @@
    %0A            
-//
 values.add(%22henr
@@ -2324,34 +2324,32 @@
 %22);%0A            
-//
 fieldsList.add(%22
@@ -2386,18 +2386,16 @@
         
-//
 values.a
@@ -2419,18 +2419,16 @@
         
-//
 fieldsLi
@@ -2471,36 +2471,14 @@
 urn 
-new Values(%22henry%22, %22smith%22)
+values
 ;%0A  
@@ -2500,28 +2500,24 @@
   @Override%0A
-    
         publ
@@ -2580,31 +2580,18 @@
 lds(
-%22
 fi
-rstname%22, %22lastname%22
+eldsList
 );%0A 

fix spelling...