/*
 *  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 java.util.Arrays;
import java.util.List;

public class AvroScheme implements Scheme {

        private static final long serialVersionUID = -2990121166902741545L;

        @Override
        public List<Object> deserialize(byte[] bytes) {
            String msg = Arrays.toString(bytes);
            return new Values(msg);
        }

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

@@ -979,68 +979,573 @@
 ing 
-msg = Arrays.toString(bytes);%0A            return new Values(
+protocolVersion = Byte.toString(bytes%5B0%5D);%0A            String mutationTypeIdx = Byte.toString(bytes%5B1%5D);%0A            String mutationType = %22%22;%0A            switch (mutationTypeIdx) %7B%0A                case %221%22:   mutationType = %22insert%22;%0A                case %222%22:   mutationType = %22update%22;%0A                case %223%22:   mutationType = %22delete%22;%0A            %7D%0A            String msg = %22%22;%0A            for (Byte theByte : bytes) %7B%0A                msg = msg + Byte.toString(theByte);%0A            %7D%0A            %0A            return new Values(protocolVersion, mutationType, 
 msg)
@@ -1617,24 +1617,24 @@
 tFields() %7B%0A
-
             
@@ -1651,16 +1651,35 @@
  Fields(
+%22version%22, %22type%22, 
 %22msg%22);%0A

convert the first two bytes