[BACKLOG-14866] Adding serializers for metrics and data. Extracted co…

…mmon code to base class. Unit tests added

pdi-engine/api/src/main/java/org/pentaho/di/engine/api/events/BaseEvent.java

@@ -45,6 +45,28 @@ public BaseEvent( S source, D data ) {
   @Override public D getData() {
     return data;
   }
+
+  @Override public boolean equals( Object o ) {
+    if ( this == o ) {
+      return true;
+    }
+    if ( !( o instanceof BaseEvent ) ) {
+      return false;
+    }
+
+    BaseEvent<?, ?> baseEvent = (BaseEvent<?, ?>) o;
+
+    if ( !source.getId().equals( baseEvent.source.getId() ) ) {
+      return false;
+    }
+    return data.equals( baseEvent.data );
+  }
+
+  @Override public int hashCode() {
+    int result = source.getId().hashCode();
+    result = 31 * result + data.hashCode();
+    return result;
+  }
 }
 
 

pdi-engine/api/src/main/java/org/pentaho/di/engine/api/events/DataEvent.java

@@ -39,8 +39,4 @@ public DataEvent( S source, Rows rows ) {
     super( source, rows );
   }
 
-
-
-
-
 }

pdi-engine/api/src/main/java/org/pentaho/di/engine/api/model/Row.java

@@ -50,8 +50,6 @@ public interface Row extends Serializable {
   List<String> getColumnNames();
 
   List<Class> getColumnTypes();
-
-
   Optional<String> getString( int index ) throws RowException;
 
   Optional<Long> getLong( int index ) throws RowException;
@@ -85,4 +83,9 @@ public interface Row extends Serializable {
   Optional<Object> getObject( String name ) throws RowException;
 
   Optional<Object[]> getObjects();
+
+  Optional<Integer> getInteger( int index ) throws RowException;
+
+  Optional<Integer> getInteger( String name) throws RowException;
+
 }

pdi-engine/api/src/main/java/org/pentaho/di/engine/api/model/Rows.java

@@ -107,14 +107,6 @@ public Rows( List<Row> rowList, TYPE type, STATE state ) {
     rows.clear();
   }
 
-  @Override public boolean equals( Object o ) {
-    return rows.equals( o );
-  }
-
-  @Override public int hashCode() {
-    return rows.hashCode();
-  }
-
   @Override public Row get( int index ) {
     return rows.get( index );
   }
@@ -159,5 +151,29 @@ public STATE getState() {
     return state;
   }
 
+  @Override public boolean equals( Object o ) {
+    if ( this == o ) {
+      return true;
+    }
+    if ( !( o instanceof Rows ) ) {
+      return false;
+    }
+
+    Rows rows1 = (Rows) o;
+
+    if ( !rows.equals( rows1.rows ) ) {
+      return false;
+    }
+    if ( state != rows1.state ) {
+      return false;
+    }
+    return type == rows1.type;
+  }
 
+  @Override public int hashCode() {
+    int result = rows.hashCode();
+    result = 31 * result + state.hashCode();
+    result = 31 * result + type.hashCode();
+    return result;
+  }
 }

pdi-engine/api/src/main/java/org/pentaho/di/engine/api/reporting/Metrics.java

@@ -93,4 +93,34 @@ public Metrics add( Metrics right ) {
       getInFlight() + right.getInFlight()
     );
   }
+
+  @Override public boolean equals( Object o ) {
+    if ( this == o ) {
+      return true;
+    }
+    if ( !( o instanceof Metrics ) ) {
+      return false;
+    }
+
+    Metrics metrics = (Metrics) o;
+
+    if ( in != metrics.in ) {
+      return false;
+    }
+    if ( out != metrics.out ) {
+      return false;
+    }
+    if ( dropped != metrics.dropped ) {
+      return false;
+    }
+    return inFlight == metrics.inFlight;
+  }
+
+  @Override public int hashCode() {
+    int result = (int) ( in ^ ( in >>> 32 ) );
+    result = 31 * result + (int) ( out ^ ( out >>> 32 ) );
+    result = 31 * result + (int) ( dropped ^ ( dropped >>> 32 ) );
+    result = 31 * result + (int) ( inFlight ^ ( inFlight >>> 32 ) );
+    return result;
+  }
 }

pdi-engine/serializers/pom.xml

@@ -74,6 +74,33 @@
       <version>2.5.3</version>
     </dependency>
 
+    <dependency>
+      <groupId>commons-codec</groupId>
+      <artifactId>commons-codec</artifactId>
+      <version>1.10</version>
+    </dependency>
+
+
+    <dependency>
+      <groupId>org.hamcrest</groupId>
+      <artifactId>java-hamcrest</artifactId>
+      <version>${hamcrest.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-all</artifactId>
+      <version>${mockito.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <version>${junit.version}</version>
+      <scope>test</scope>
+    </dependency>
+
+
   </dependencies>
 
   <build>

pdi-engine/serializers/src/main/java/org/pentaho/pdi/engine/serializers/BaseSerializer.java

@@ -0,0 +1,67 @@
+/*
+ * *****************************************************************************
+ *
+ *  Pentaho Data Integration
+ *
+ *  Copyright (C) 2002-2017 by Pentaho : http://www.pentaho.com
+ *
+ *  *******************************************************************************
+ *  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 org.pentaho.pdi.engine.serializers;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.pentaho.di.engine.api.events.PDIEvent;
+import org.pentaho.osgi.objecttunnel.TunnelSerializer;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Created by nbaker on 3/4/17.
+ */
+abstract class BaseSerializer<T extends PDIEvent> implements TunnelSerializer<T> {
+
+  ObjectMapper mapper = new ObjectMapper();
+  private Class<T> tClass;
+
+  public BaseSerializer( Class<T> tClass ) {
+    this.tClass = tClass;
+  }
+
+  @Override public List<Class> getSupportedClasses() {
+    return Collections.singletonList( tClass );
+  }
+
+  @Override public String serialize( Object object ) {
+    try {
+      return mapper.writer().writeValueAsString( object );
+    } catch ( JsonProcessingException e ) {
+      throw new RuntimeException( e );
+    }
+  }
+
+  @Override public T deserialize( String serializedString ) {
+    try {
+      return mapper.readValue( serializedString, tClass );
+    } catch ( IOException e ) {
+      throw new RuntimeException( e );
+    }
+  }
+}