From f2e8a5d8068dce478b0287ca54d401af96e268f0 Mon Sep 17 00:00:00 2001
From: Aaron Jacobs <jacobsa@google.com>
Date: Fri, 24 Jul 2015 15:04:28 +1000
Subject: [PATCH] Connection.readMessage

---
 connection.go                 | 24 +++++++++++++++++++++---
 internal/buffer/in_message.go | 11 +++++++++++
 2 files changed, 32 insertions(+), 3 deletions(-)

diff --git a/connection.go b/connection.go
index d81fff8..c48e79f 100644
--- a/connection.go
+++ b/connection.go
@@ -15,7 +15,6 @@
 package fuse
 
 import (
-	"errors"
 	"fmt"
 	"log"
 	"path"
@@ -215,8 +214,27 @@ func (c *Connection) destroyInMessage(m *buffer.InMessage) {
 // Read the next message from the kernel. The message must later be destroyed
 // using destroyInMessage.
 func (c *Connection) readMessage() (m *buffer.InMessage, err error) {
-	err = errors.New("TODO")
-	return
+	// Allocate a message.
+	m = c.allocateInMessage()
+
+	// Loop past transient errors.
+	for {
+		// Lock and read.
+		//
+		// TODO(jacobsa): Ensure that we document concurrency constraints that make
+		// it safe, then kill the lock here.
+		c.wrapped.Rio.RLock()
+		err = m.Init(c.wrapped.Dev)
+		c.wrapped.Rio.RUnlock()
+
+		if err != nil {
+			c.destroyInMessage(m)
+			m = nil
+			return
+		}
+
+		return
+	}
 }
 
 // Read the next op from the kernel process. Return io.EOF if the kernel has
diff --git a/internal/buffer/in_message.go b/internal/buffer/in_message.go
index ee836fc..72ea5c3 100644
--- a/internal/buffer/in_message.go
+++ b/internal/buffer/in_message.go
@@ -55,11 +55,22 @@ func (m *InMessage) Init(r io.Reader) (err error) {
 		return
 	}
 
+	// Make sure the message is long enough that calling Header is safe.
 	if uintptr(n) < unsafe.Sizeof(fusekernel.InHeader{}) {
 		err = fmt.Errorf("Unexpectedly read only %d bytes.", n)
 		return
 	}
 
+	// Check the header's length.
+	if int(m.Header().Len) != n {
+		err = fmt.Errorf(
+			"Header says %d bytes, but we read %d",
+			m.Header().Len,
+			n)
+
+		return
+	}
+
 	m.remaining = m.storage[:n]
 	return
 }