add commit message

Author: majecty

core/src/consensus/tendermint/message.rs

@@ -79,6 +79,8 @@ const MESSAGE_ID_PROPOSAL_BLOCK: u8 = 0x02;
 const MESSAGE_ID_STEP_STATE: u8 = 0x03;
 const MESSAGE_ID_REQUEST_MESSAGE: u8 = 0x04;
 const MESSAGE_ID_REQUEST_PROPOSAL: u8 = 0x05;
+const MESSAGE_ID_REQUEST_COMMIT: u8 = 0x06;
+const MESSAGE_ID_COMMIT: u8 = 0x07;
 
 #[derive(Debug, PartialEq)]
 pub enum TendermintMessage {
@@ -102,6 +104,14 @@ pub enum TendermintMessage {
         height: Height,
         view: View,
     },
+    RequestCommit {
+        height: Height,
+    },
+    Commit {
+        height: Height,
+        block: Bytes,
+        votes: Vec<ConsensusMessage>,
+    },
 }
 
 impl Encodable for TendermintMessage {
@@ -160,6 +170,24 @@ impl Encodable for TendermintMessage {
                 s.append(height);
                 s.append(view);
             }
+            TendermintMessage::RequestCommit {
+                height,
+            } => {
+                s.begin_list(2);
+                s.append(&MESSAGE_ID_REQUEST_COMMIT);
+                s.append(height);
+            }
+            TendermintMessage::Commit {
+                height,
+                block,
+                votes,
+            } => {
+                s.begin_list(4);
+                s.append(&MESSAGE_ID_COMMIT);
+                s.append(height);
+                s.append(block);
+                s.append_list(votes);
+            }
         }
     }
 }
@@ -253,6 +281,36 @@ impl Decodable for TendermintMessage {
                     view,
                 }
             }
+            MESSAGE_ID_REQUEST_COMMIT => {
+                let item_count = rlp.item_count()?;
+                if item_count != 2 {
+                    return Err(DecoderError::RlpIncorrectListLen {
+                        got: item_count,
+                        expected: 2,
+                    })
+                }
+                let height = rlp.at(1)?.as_val()?;
+                TendermintMessage::RequestCommit {
+                    height,
+                }
+            }
+            MESSAGE_ID_COMMIT => {
+                let item_count = rlp.item_count()?;
+                if item_count != 4 {
+                    return Err(DecoderError::RlpIncorrectListLen {
+                        got: item_count,
+                        expected: 4,
+                    })
+                }
+                let height = rlp.at(1)?.as_val()?;
+                let block = rlp.at(2)?.as_val()?;
+                let votes = rlp.at(3)?.as_list()?;
+                TendermintMessage::Commit {
+                    height,
+                    block,
+                    votes,
+                }
+            }
             _ => return Err(DecoderError::Custom("Unknown message id detected")),
         })
     }

core/src/consensus/tendermint/network.rs

@@ -352,6 +352,45 @@ impl NetworkExtension<Event> for TendermintExtension {
                     self.send_votes(token, votes);
                 }
             }
+            Ok(TendermintMessage::RequestCommit {
+                height,
+            }) => {
+                ctrace!(ENGINE, "Received RequestCommit for {} from {:?}", height, token);
+                let (result, receiver) = crossbeam::unbounded();
+                self.inner
+                    .send(worker::Event::RequestCommit {
+                        height,
+                        result,
+                    })
+                    .unwrap();
+
+                if let Ok(message) = receiver.recv() {
+                    ctrace!(ENGINE, "Send commit for {} to {:?}", height, token);
+                    self.api.send(token, Arc::new(message));
+                }
+            }
+            Ok(TendermintMessage::Commit {
+                height,
+                block,
+                votes,
+            }) => {
+                ctrace!(ENGINE, "Received Commit for {:?} from {:?}", height, token);
+                let (result, receiver) = crossbeam::unbounded();
+                self.inner
+                    .send(worker::Event::GetCommit {
+                        height,
+                        block: block.clone(),
+                        votes,
+                        result,
+                    })
+                    .unwrap();
+
+                if let Some(c) = receiver.recv().unwrap() {
+                    if let Err(e) = c.import_block(block) {
+                        cinfo!(ENGINE, "Failed to import committed block {:?}", e);
+                    }
+                }
+            }
             _ => cinfo!(ENGINE, "Invalid message from peer {}", token),
         }
     }