Replace uses of bitcoin::consensus with consensus_encoding

diff --git a/bitcoin/src/bip158.rs b/bitcoin/src/bip158.rs
index dcb5cdce..87bb4038 100644
--- a/bitcoin/src/bip158.rs
+++ b/bitcoin/src/bip158.rs
@@ -44,7 +44,7 @@ use internals::array::ArrayExt as _;
 use io::{BufRead, Write};
 
 use crate::block::{Block, BlockHash, Checked};
-use crate::consensus::{ReadExt, WriteExt};
+use crate::encoding::{CompactSizeEncoder, CompactSizeU64Decoder, ExactSizeEncoder as _};
 use crate::prelude::{BTreeSet, Borrow, Vec};
 use crate::script::{ScriptPubKey, ScriptPubKeyExt as _};
 use crate::transaction::OutPoint;
@@ -220,7 +220,8 @@ impl GcsFilterReader {
         I::Item: Borrow<[u8]>,
         R: BufRead + ?Sized,
     {
-        let n_elements = reader.read_compact_size().map_err(Error::InvalidCompactSize)?;
+        let n_elements = io::decode_from_read_with::<CompactSizeU64Decoder, _>(&mut *reader)
+            .map_err(Error::InvalidCompactSize)?;
         // map hashes to [0, n_elements << grp]
         let nm = n_elements * self.m;
         let mut mapped =
@@ -263,7 +264,8 @@ impl GcsFilterReader {
         I::Item: Borrow<[u8]>,
         R: BufRead + ?Sized,
     {
-        let n_elements = reader.read_compact_size().map_err(Error::InvalidCompactSize)?;
+        let n_elements = io::decode_from_read_with::<CompactSizeU64Decoder, _>(&mut *reader)
+            .map_err(Error::InvalidCompactSize)?;
         // map hashes to [0, n_elements << grp]
         let nm = n_elements * self.m;
         let mut mapped =
@@ -338,7 +340,9 @@ impl<'a, W: Write> GcsFilterWriter<'a, W> {
         mapped.sort_unstable();
 
         // write number of elements as varint
-        let mut wrote = self.writer.emit_compact_size(mapped.len())?;
+        let mut encoder = CompactSizeEncoder::new(mapped.len());
+        let mut wrote = encoder.len();
+        io::drain_to_writer(&mut encoder, &mut self.writer)?;
 
         // write out deltas of sorted values into a Golomb-Rice coded bit stream
         let mut writer = BitStreamWriter::new(self.writer);
@@ -501,7 +505,6 @@ pub mod error {
 
     use internals::write_err;
 
-    use crate::consensus;
     use crate::transaction::OutPoint;
 
     /// Errors for blockfilter.
@@ -511,7 +514,7 @@ pub mod error {
         /// Missing UTXO, cannot calculate script filter.
         UtxoMissing(OutPoint),
         /// Invalid CompactSize encoded element count in the filter.
-        InvalidCompactSize(consensus::Error),
+        InvalidCompactSize(io::ReadError<encoding::CompactSizeDecoderError>),
         /// I/O error reading or writing binary serialization of the filter.
         Io(io::Error),
     }
@@ -697,33 +700,35 @@ mod test {
 
     #[test]
     fn malformed_filter_count_errors() {
-        use crate::consensus::Error::Parse as ConsensusParse;
-        use crate::consensus::ParseError::{MissingData, NonMinimalCompactSize};
+        // Check the inner private error type via text
+        fn assert_error_string(result: Result<bool, Error>, substring: &str) {
+            match result {
+                Err(Error::InvalidCompactSize(io::ReadError::Decode(c_err))) => {
+                    let err_msg = format!("{}", c_err);
+                    assert!(err_msg.contains(substring));
+                }
+                _ => panic!("Incorrect error type: {:?}", result),
+            }
+        }
 
         let query = [hex!("000000")];
         let reader = GcsFilterReader::new(0, 0, M, P);
 
         let mut bytes = &[0xfd][..];
         let result = reader.match_any(&mut bytes, query.iter().map(|v| v.as_slice()));
-        assert!(matches!(result, Err(Error::InvalidCompactSize(ConsensusParse(MissingData)))));
+        assert_error_string(result, "required at least");
 
         let mut bytes = &[0xfd][..];
         let result = reader.match_all(&mut bytes, query.iter().map(|v| v.as_slice()));
-        assert!(matches!(result, Err(Error::InvalidCompactSize(ConsensusParse(MissingData)))));
+        assert_error_string(result, "required at least");
 
         let mut bytes = &[0xfd, 0xfc, 0x00][..];
         let result = reader.match_any(&mut bytes, query.iter().map(|v| v.as_slice()));
-        assert!(matches!(
-            result,
-            Err(Error::InvalidCompactSize(ConsensusParse(NonMinimalCompactSize)))
-        ));
+        assert_error_string(result, "not encoded minimally");
 
         let mut bytes = &[0xfd, 0xfc, 0x00][..];
         let result = reader.match_all(&mut bytes, query.iter().map(|v| v.as_slice()));
-        assert!(matches!(
-            result,
-            Err(Error::InvalidCompactSize(ConsensusParse(NonMinimalCompactSize)))
-        ));
+        assert_error_string(result, "not encoded minimally");
     }
 
     #[test]
diff --git a/bitcoin/src/blockdata/script/borrowed.rs b/bitcoin/src/blockdata/script/borrowed.rs
index 332d2dcf..88aaf0e7 100644
--- a/bitcoin/src/blockdata/script/borrowed.rs
+++ b/bitcoin/src/blockdata/script/borrowed.rs
@@ -10,12 +10,12 @@ use super::{
     RedeemScriptSizeError, Script, ScriptHash, ScriptHashableTag, ScriptPubKey, ScriptSig,
     TapScript, WScriptHash, WitnessScript, WitnessScriptSizeError,
 };
-use crate::consensus::Encodable;
+use crate::encoding::{Encode, ExactSizeEncoder};
 use crate::key::{LegacyPublicKey, UntweakedPublicKey, WPubkeyHash};
 use crate::opcodes::all::*;
 use crate::opcodes::{self, Opcode, OpcodeExt as _};
 use crate::policy::{DUST_RELAY_TX_FEE, MAX_OP_RETURN_RELAY};
-use crate::prelude::{sink, String, ToString};
+use crate::prelude::{String, ToString};
 use crate::script::{self, ScriptPubKeyBufExt as _};
 use crate::taproot::{LeafVersion, TapLeafHash, TapLeafHashExt as _, TapNodeHash};
 use crate::witness_program::P2A_PROGRAM;
@@ -549,11 +549,11 @@ internal_macros::define_extension_trait! {
                 } else if self.is_witness_program() {
                     32 + 4 + 1 + (107 / 4) + 4 + // The spend cost copied from Core
                     8 + // The serialized size of the TxOut's amount field
-                    self.consensus_encode(&mut sink()).expect("sinks don't error").to_u64() // The serialized size of this script_pubkey
+                    self.encoder().len().to_u64() // The serialized size of this script_pubkey
                 } else {
                     32 + 4 + 1 + 107 + 4 + // The spend cost copied from Core
                     8 + // The serialized size of the TxOut's amount field
-                    self.consensus_encode(&mut sink()).expect("sinks don't error").to_u64() // The serialized size of this script_pubkey
+                    self.encoder().len().to_u64() // The serialized size of this script_pubkey
                 })?
                 / 1000; // divide by 1000 like in Core to get value as it cancels out DEFAULT_MIN_RELAY_TX_FEE
                         // Note: We ensure the division happens at the end, since Core performs the division at the end.
diff --git a/bitcoin/src/consensus_validation.rs b/bitcoin/src/consensus_validation.rs
index 7dad714c..f91524a4 100644
--- a/bitcoin/src/consensus_validation.rs
+++ b/bitcoin/src/consensus_validation.rs
@@ -5,9 +5,9 @@
 //! Relies on the `bitcoinconsensus` crate that uses Bitcoin Core libconsensus to perform validation.
 
 use crate::amount::Amount;
-use crate::consensus::encode;
 #[cfg(doc)]
 use crate::consensus_validation;
+use crate::encoding;
 use crate::internal_macros::define_extension_trait;
 use crate::script::ScriptPubKey;
 use crate::transaction::{OutPoint, Transaction, TxOut};
@@ -98,7 +98,7 @@ where
     S: FnMut(&OutPoint) -> Option<TxOut>,
     F: Into<u32>,
 {
-    let serialized_tx = encode::serialize(tx);
+    let serialized_tx = encoding::encode_to_vec(tx);
     let flags: u32 = flags.into();
     for (idx, input) in tx.inputs.iter().enumerate() {
         if let Some(output) = spent(&input.previous_output) {
diff --git a/bitcoin/src/crypto/sighash.rs b/bitcoin/src/crypto/sighash.rs
index 75cbca0e..11fd1ffc 100644
--- a/bitcoin/src/crypto/sighash.rs
+++ b/bitcoin/src/crypto/sighash.rs
@@ -18,6 +18,7 @@ use core::str;
 use arbitrary::{Arbitrary, Unstructured};
 use crypto::key::{TweakedKeypair, UntweakedKeypair};
 use crypto::{ecdsa, taproot, PrivateKey};
+use encoding::CompactSizeEncoder;
 use hashes::{hash_newtype, sha256, sha256d, sha256t, sha256t_tag};
 use io::Write;
 
@@ -714,20 +715,18 @@ impl<R: Borrow<Transaction>> SighashCache<R> {
             script_pubkey: &crate::script::Script<T>,
             sighash_type: u32,
         ) -> Result<(), io::Error> {
-            use crate::consensus::encode::WriteExt;
-
             let (sighash, anyone_can_pay) =
                 EcdsaSighashType::from_consensus(sighash_type).split_anyonecanpay_flag();
 
             io::encode_to_writer(&self_.version, &mut writer)?;
             // Add all inputs necessary..
             if anyone_can_pay {
-                writer.emit_compact_size(1u8)?;
+                io::drain_to_writer(&mut CompactSizeEncoder::new(1), &mut writer)?;
                 io::encode_to_writer(&self_.inputs[input_index].previous_output, &mut writer)?;
                 io::encode_to_writer(script_pubkey, &mut writer)?;
                 io::encode_to_writer(&self_.inputs[input_index].sequence, &mut writer)?;
             } else {
-                writer.emit_compact_size(self_.inputs.len())?;
+                io::drain_to_writer(&mut CompactSizeEncoder::new(self_.inputs.len()), &mut writer)?;
                 for (n, input) in self_.inputs.iter().enumerate() {
                     io::encode_to_writer(&input.previous_output, &mut writer)?;
                     if n == input_index {
@@ -758,7 +757,7 @@ impl<R: Borrow<Transaction>> SighashCache<R> {
                     // sign all outputs up to and including this one, but erase
                     // all of them except for this one
                     let count = input_index.min(self_.outputs.len() - 1);
-                    writer.emit_compact_size(count + 1)?;
+                    io::drain_to_writer(&mut CompactSizeEncoder::new(count + 1), &mut writer)?;
                     for _ in 0..count {
                         // consensus encoding of the "NULL txout" - max amount, empty script_pubkey
                         writer
@@ -767,7 +766,7 @@ impl<R: Borrow<Transaction>> SighashCache<R> {
                     io::encode_to_writer(&self_.outputs[count], &mut writer)?;
                 }
                 EcdsaSighashType::None => {
-                    writer.emit_compact_size(0u8)?;
+                    io::drain_to_writer(&mut CompactSizeEncoder::new(0), &mut writer)?;
                 }
                 _ => unreachable!(),
             };
diff --git a/bitcoin/src/sign_message.rs b/bitcoin/src/sign_message.rs
index 9183cdba..82d1b5e1 100644
--- a/bitcoin/src/sign_message.rs
+++ b/bitcoin/src/sign_message.rs
@@ -5,9 +5,9 @@
 //! This module provides signature related functions including secp256k1 signature recovery when
 //! library is used with the `secp-recovery` feature.
 
+use encoding::CompactSizeEncoder;
 use hashes::{sha256d, HashEngine};
 
-use crate::consensus::encode::WriteExt;
 #[cfg(feature = "secp-recovery")]
 use crate::key::PrivateKeyExt as _;
 #[cfg(feature = "secp-recovery")]
@@ -169,7 +169,7 @@ pub fn signed_msg_hash(msg: impl AsRef<[u8]>) -> sha256d::Hash {
     let msg_bytes = msg.as_ref();
     let mut engine = sha256d::Hash::engine();
     engine.input(BITCOIN_SIGNED_MSG_PREFIX);
-    engine.emit_compact_size(msg_bytes.len()).expect("engines don't error");
+    hashes::drain_to_engine(&mut CompactSizeEncoder::new(msg_bytes.len()), &mut engine);
     engine.input(msg_bytes);
     sha256d::Hash::from_engine(engine)
 }
diff --git a/bitcoin/src/taproot/mod.rs b/bitcoin/src/taproot/mod.rs
index 24ece6c5..15a5ecb6 100644
--- a/bitcoin/src/taproot/mod.rs
+++ b/bitcoin/src/taproot/mod.rs
@@ -21,7 +21,6 @@ use io::Write;
 #[cfg(feature = "serde")]
 use serde::Deserialize;
 
-use crate::consensus::Encodable;
 use crate::crypto::key::{
     SerializedXOnlyPublicKey, TapTweak, TweakedPublicKey, UntweakedPublicKey,
 };
@@ -69,8 +68,8 @@ crate::internal_macros::define_extension_trait! {
         /// Computes the leaf hash from components.
         fn from_script(script: &TapScript, ver: LeafVersion) -> Self {
             let mut eng = sha256t::Hash::<TapLeafTag>::engine();
-            ver.to_consensus().consensus_encode(&mut eng).expect("engines don't error");
-            script.consensus_encode(&mut eng).expect("engines don't error");
+            eng.input(&[ver.to_consensus()]);
+            hashes::encode_to_engine(script, &mut eng);
             let inner = sha256t::Hash::<TapLeafTag>::from_engine(eng);
             Self::from_byte_array(inner.to_byte_array())
         }