fix produce proto type + error (#142)

Author: shay23b

memphis/producer.py

@@ -52,6 +52,9 @@ async def validate_msg(self, message):
             message = str(msg.loc.source.body)
             message = message.encode("utf-8")
             return message
+        elif hasattr(message, "SerializeToString"):
+            msgToSend = message.SerializeToString()
+            return msgToSend
         elif not isinstance(message, bytearray) and not isinstance(message, dict):
             raise MemphisSchemaError("Unsupported message type")
         else:
@@ -78,6 +81,13 @@ def validate_protobuf(self, message):
                 msgToSend = message.SerializeToString()
                 proto_msg.ParseFromString(msgToSend)
                 proto_msg.SerializeToString()
+                try:
+                    proto_msg.ParseFromString(msgToSend)
+                    proto_msg.SerializeToString()
+                except Exception as e:
+                    if "parsing message" in str(e):
+                        e = "Error parsing protobuf message"
+                    raise MemphisSchemaError(str(e))
                 return msgToSend
 
             else: