READEME update and throw error when previously ingested data format m…

…ismatch (#128)

README.md

@@ -362,9 +362,13 @@ The consumer will try to fetch messages every ```pullInterval``` (that was given
 ```go
 func handler(msgs []*memphis.Msg, err error, ctx context.Context) {
 	if err != nil {
-		m := msgs[0]
-		fmt.Println(string(m.Data()))
-		m.Ack()
+		fmt.Printf("Fetch failed: %v", err)
+		return
+	}
+
+	for _, msg := range msgs {
+		fmt.Println(string(msg.Data()))
+		msg.Ack()
 	}
 }
 
@@ -373,6 +377,25 @@ consumer.Consume(handler,
 )
 ```
 
+#### Consumer schema deserialization
+To get messages deserialized, use `msg.DataDeserialized()`.  
+
+```go
+func handler(msgs []*memphis.Msg, err error, ctx context.Context) {
+	if err != nil {
+		fmt.Printf("Fetch failed: %v", err)
+		return
+	}
+
+	for _, msg := range msgs {
+		fmt.Println(string(msg.DataDeserialized()))
+		msg.Ack()
+	}
+}
+```
+
+if you have ingested data into station in one format, afterwards you apply a schema on the station, the consumer won't deserialize the previously ingested data. For example, you have ingested string into the station and attached a protobuf schema on the station. In this case, consumer won't deserialize the string.
+
 ### Fetch a single batch of messages
 ```go
 msgs, err := conn.FetchMessages("<station-name>", "<consumer-name>",

consumer.go

@@ -103,6 +103,11 @@ func (m *Msg) DataDeserialized() (any, error) {
 
 	msgBytes := m.msg.Data
 
+	_, err = sd.validateMsg(msgBytes)
+	if err != nil {
+		return nil, memphisError(errors.New("Deserialization has been failed since the message format does not align with the currently attached schema: " + err.Error()))
+	}
+
 	switch sd.schemaType {
 	case "protobuf":
 		pMsg := dynamicpb.NewMessage(sd.msgDescriptor)