Fix warnings and lints

.gitignore

@@ -2,3 +2,4 @@ target/
 **/certs/
 .DS_Store
 **/logs/
+log/path

log/benches/read_benchmark.rs

@@ -45,8 +45,8 @@ async fn read_records(path: impl AsRef<Path>) {
 
         match slice.messages().as_mut() {
             Some(ref mut iterator) => {
-                while let Some(next) = iterator.next().await {
-                    black_box(next).unwrap();
+                while let Some(next) = iterator.next().await.unwrap() {
+                    black_box(next);
                 }
             }
             None => break,

log/benches/write_benchmark.rs

@@ -27,7 +27,7 @@ async fn log_task() {
     let log = MessageLog::open(Arc::new(config)).await.unwrap();
 
     for _ in 0..NUM_OF_MESSAGES {
-        let batch = Bytes::copy_from_slice(&vec![1; 32]);
+        let batch = Bytes::copy_from_slice(&[1; 32]);
         let message = Message::single(&batch, 1);
         log.write(message).await.unwrap();
     }
@@ -38,7 +38,7 @@ async fn log_task() {
 pub fn benchmark(c: &mut Criterion) {
     c.bench_function("write 1_000_000 records", |b| {
         let runtime = tokio::runtime::Runtime::new().expect("Failed to construct executor");
-        b.to_async(runtime).iter(|| log_task());
+        b.to_async(runtime).iter(log_task);
     });
 }
 

log/tests/helpers.rs

@@ -48,7 +48,7 @@ impl TestWrapper {
 
     pub async fn write_records(&mut self, records: &[String]) {
         for record in records {
-            self.write(&record).await;
+            self.write(record).await;
         }
     }
 

protocol/src/codec.rs

@@ -143,7 +143,7 @@ mod tests {
 
         let mut codec = MessageCodec;
         let mut buffer = BytesMut::new();
-        let expected = Bytes::from_static(b"\0\0\0\0\0\0\06\x04\x01\x01\0\0\0\0\0\0\0\x04\0\0\0\0\0\0\0test\x06\0\0\0\0\0\0\0header\x0b\0\0\0\0\0\0\0Hello world");
+        let expected = Bytes::from_static(b"\0\0\0\0\0\0\x006\x04\x01\x01\0\0\0\0\0\0\0\x04\0\0\0\0\0\0\0test\x06\0\0\0\0\0\0\0header\x0b\0\0\0\0\0\0\0Hello world");
 
         codec.encode(frame, &mut buffer).unwrap();
 
@@ -279,7 +279,7 @@ mod tests {
     #[test]
     fn decodes_message_frame_with_header() {
         let mut codec = MessageCodec;
-        let mut src = BytesMut::from("\0\0\0\0\0\0\06\x04\x01\x01\0\0\0\0\0\0\0\x04\0\0\0\0\0\0\0test\x06\0\0\0\0\0\0\0header\x0b\0\0\0\0\0\0\0Hello world");
+        let mut src = BytesMut::from("\0\0\0\0\0\0\x006\x04\x01\x01\0\0\0\0\0\0\0\x04\0\0\0\0\0\0\0test\x06\0\0\0\0\0\0\0header\x0b\0\0\0\0\0\0\0Hello world");
 
         let mut h = HashMap::new();
         h.insert("test".to_owned(), "header".to_owned());

server/src/sink/fanout_many.rs

@@ -44,10 +44,10 @@ impl<K, V> FanoutMany<K, V> {
         ret
     }
 
-    pub fn remove<Q: ?Sized>(&mut self, k: &Q) -> Option<V>
+    pub fn remove<Q>(&mut self, k: &Q) -> Option<V>
     where
         K: Borrow<Q>,
-        Q: Hash + Eq,
+        Q: Hash + Eq + ?Sized,
     {
         for i in 0..self.entries.len() {
             if self.entries[i].0.borrow() == k {

tests/tests/streams/request_reply.rs

@@ -8,7 +8,7 @@ use uuid::Uuid;
 #[tokio::test]
 async fn request_reply_successful() -> Result<()> {
     let client = TestClient::start().await?;
-    let _ = client.start_replier(None);
+    client.start_replier(None);
 
     let mut requestor = client.requestor(None).await?;
     let reply = requestor.request(Request::Ping).await?;
@@ -21,7 +21,7 @@ async fn request_reply_successful() -> Result<()> {
 #[tokio::test]
 async fn request_fails_if_exceeds_timeout() -> Result<()> {
     let client = TestClient::start().await?;
-    let _ = client.start_replier(Some(Duration::from_secs(3)));
+    client.start_replier(Some(Duration::from_secs(3)));
 
     let mut requestor = client.requestor(Some(Duration::from_secs(2))).await?;
     let reply = requestor.request(Request::Ping).await;
@@ -34,7 +34,7 @@ async fn request_fails_if_exceeds_timeout() -> Result<()> {
 #[tokio::test]
 async fn concurrent_requests_are_routed_successfully() -> Result<()> {
     let client = TestClient::start().await?;
-    let _ = client.start_replier(None);
+    client.start_replier(None);
 
     let requestor = client.requestor(None).await?;
     let mut tasks = vec![];