/*
 *  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 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 values;
        }

        @Override
        public Fields getOutputFields() {
            return new Fields(fieldsList);
        }
}

@@ -1197,76 +1197,8 @@
    %0A
-    private List%3CString%3E fieldsList = new ArrayList%3CString%3E();%0A    %0A
     
@@ -1528,18 +1528,16 @@
         
-/*
  String 
@@ -1893,17 +1893,10 @@
  // 
-FirstName
+Id
 %0A   
@@ -1916,37 +1916,148 @@
     
-fieldsList.add(%22firstname
+values.add(AvroSchemaUtils.getIntegerValueByKey(insertMutation.getIntegers(), %22id
 %22)
+)
 ;%0A
+                    %0A                    // FirstName%0A
     
@@ -2159,24 +2159,25 @@
 rstname%22));%0A
+%0A
             
@@ -2188,88 +2188,127 @@
     
-%0A                %7D catch(IOException e) %7B%0A                    e.printStackTrace(
+// LastName%0A                    values.add(AvroSchemaUtils.getStringValueByKey(insertMutation.getStrings(), %22lastname%22)
 );%0A 
@@ -2314,34 +2314,36 @@
                 
-%7D%0A
+    
 %0A            %7D*/
@@ -2343,99 +2343,82 @@
     
-%7D*/%0A            %0A            values.add(%22henry%22);%0A            fieldsList.add(%22firstname%22);%0A
+    %7D catch(IOException e) %7B%0A                    e.printStackTrace();%0A    
     
@@ -2417,32 +2417,34 @@
                 
+%7D%0A
 %0A            val
@@ -2444,69 +2444,22 @@
     
-values.add(%22smith%22);%0A            fieldsList.add(%22lastname%22);%0A
+%7D%0A            
 %0A   
@@ -2587,18 +2587,37 @@
 lds(
-fieldsList
+%22id%22, %22firstname%22, %22lastname%22
 );%0A 

fix scheme