AJ-655: modify when send event, not serialize decimal scale;

Author: chengyitian

src/com/xxdb/data/BasicDecimal128.java

@@ -86,6 +86,30 @@ public void writeScalarToOutputStream(ExtendedDataOutput out) throws IOException
         out.writeBigIntArray(newArray, 0, newArray.length);
     }
 
+    public void writeScalarRawDataToOutputStream(ExtendedDataOutput out) throws IOException {
+        // not write scale
+
+        byte[] newArray = new byte[16];
+        byte[] originalArray = unscaledValue.toByteArray();
+
+        if(originalArray.length == 0 || originalArray.length > 16){
+            throw new RuntimeException("byte length of Decimal128 "+originalArray.length+" cannot be less than 0 or exceed 16.");
+        }
+
+        if (originalArray[0] >= 0) {
+            // if first bit is 0, represent non-negative.
+            System.arraycopy(originalArray, 0, newArray, 16 - originalArray.length, originalArray.length);
+        } else {
+            // if first bit is 1, represent negative.
+            System.arraycopy(originalArray, 0, newArray, 16 - originalArray.length, originalArray.length);
+            for (int i = 0; i < 16 - originalArray.length; i++) {
+                newArray[i] = -1;
+            }
+        }
+
+        out.writeBigIntArray(newArray, 0, newArray.length);
+    }
+
     public static void reverseByteArray(byte[] array) {
         int left = 0;
         int right = array.length - 1;

src/com/xxdb/data/BasicDecimal32.java

@@ -75,6 +75,11 @@ public void writeScalarToOutputStream(ExtendedDataOutput out) throws IOException
         out.writeInt(value_);
     }
 
+    public void writeScalarRawDataToOutputStream(ExtendedDataOutput out) throws IOException {
+        // not write scale
+        out.writeInt(value_);
+    }
+
     @Override
     public DATA_CATEGORY getDataCategory() {
         return DATA_CATEGORY.DENARY;

src/com/xxdb/data/BasicDecimal64.java

@@ -76,6 +76,11 @@ public void writeScalarToOutputStream(ExtendedDataOutput out) throws IOException
         out.writeLong(value_);
     }
 
+    public void writeScalarRawDataToOutputStream(ExtendedDataOutput out) throws IOException {
+        // not write scale
+        out.writeLong(value_);
+    }
+
     @Override
     public DATA_CATEGORY getDataCategory() {
         return DATA_CATEGORY.DENARY;

src/com/xxdb/streaming/client/cep/ScalarAttributeSerializer.java

@@ -1,7 +1,6 @@
 package com.xxdb.streaming.client.cep;
 
-import com.xxdb.data.AbstractScalar;
-import com.xxdb.data.Entity;
+import com.xxdb.data.*;
 import com.xxdb.io.ExtendedDataOutput;
 import java.io.IOException;
 
@@ -16,7 +15,15 @@ public ScalarAttributeSerializer(int unitLen) {
 
     @Override
     public void serialize(Entity attribute, ExtendedDataOutput out) throws IOException {
-        ((AbstractScalar)attribute).writeScalarToOutputStream(out);
+        if (attribute.getDataType() == Entity.DATA_TYPE.DT_DECIMAL32)
+            ((BasicDecimal32) attribute).writeScalarRawDataToOutputStream(out);
+        else if (attribute.getDataType() == Entity.DATA_TYPE.DT_DECIMAL64) {
+            ((BasicDecimal64) attribute).writeScalarRawDataToOutputStream(out);
+        } else if (attribute.getDataType() == Entity.DATA_TYPE.DT_DECIMAL128) {
+            ((BasicDecimal128) attribute).writeScalarRawDataToOutputStream(out);
+        } else {
+            ((AbstractScalar)attribute).writeScalarToOutputStream(out);
+        }
     }
 
 }