diff --git a/runtime/internal/flow/conn/conn.go b/runtime/internal/flow/conn/conn.go
index 16ece68..6f440e8 100644
--- a/runtime/internal/flow/conn/conn.go
+++ b/runtime/internal/flow/conn/conn.go
@@ -12,6 +12,10 @@
 	"v.io/v23/security"
 )
 
+// flowID is a number assigned to identify a flow.
+// Each flow on a given conn will have a unique number.
+type flowID uint64
+
 // FlowHandlers process accepted flows.
 type FlowHandler interface {
 	// HandleFlow processes an accepted flow.
diff --git a/runtime/internal/flow/conn/errors.vdl b/runtime/internal/flow/conn/errors.vdl
new file mode 100644
index 0000000..253ec86
--- /dev/null
+++ b/runtime/internal/flow/conn/errors.vdl
@@ -0,0 +1,19 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package conn
+
+// These messages are constructed so as to avoid embedding a component/method name
+// and are thus more suitable for inclusion in other verrors.
+// This practice of omitting {1}{2} should be used throughout the flow implementations
+// since all of their errors are intended to be used as arguments to higher level errors.
+// TODO(suharshs,toddw): Allow skipping of {1}{2} in vdl generated errors.
+error (
+  InvalidMsg(typ byte, size, field int64) {
+  "en":"message of type{:typ} and size{:size} failed decoding at field{:field}."}
+  InvalidControlMsg(cmd byte, size, field int64) {
+  "en":"control message of cmd{:cmd} and size{:size} failed decoding at field{:field}."}
+  UnknownMsg(typ byte) {"en":"unknown message type{:typ}."}
+  UnknownControlMsg(cmd byte) {"en":"unknown control command{:cmd}."}
+)
diff --git a/runtime/internal/flow/conn/errors.vdl.go b/runtime/internal/flow/conn/errors.vdl.go
new file mode 100644
index 0000000..15fb4cd
--- /dev/null
+++ b/runtime/internal/flow/conn/errors.vdl.go
@@ -0,0 +1,49 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// This file was auto-generated by the vanadium vdl tool.
+// Source: errors.vdl
+
+package conn
+
+import (
+	// VDL system imports
+	"v.io/v23/context"
+	"v.io/v23/i18n"
+	"v.io/v23/verror"
+)
+
+var (
+	ErrInvalidMsg        = verror.Register("v.io/x/ref/runtime/internal/flow/conn.InvalidMsg", verror.NoRetry, "{1:}{2:} message of type{:3} and size{:4} failed decoding at field{:5}.")
+	ErrInvalidControlMsg = verror.Register("v.io/x/ref/runtime/internal/flow/conn.InvalidControlMsg", verror.NoRetry, "{1:}{2:} control message of cmd{:3} and size{:4} failed decoding at field{:5}.")
+	ErrUnknownMsg        = verror.Register("v.io/x/ref/runtime/internal/flow/conn.UnknownMsg", verror.NoRetry, "{1:}{2:} unknown message type{:3}.")
+	ErrUnknownControlMsg = verror.Register("v.io/x/ref/runtime/internal/flow/conn.UnknownControlMsg", verror.NoRetry, "{1:}{2:} unknown control command{:3}.")
+)
+
+func init() {
+	i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrInvalidMsg.ID), "{1:}{2:} message of type{:3} and size{:4} failed decoding at field{:5}.")
+	i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrInvalidControlMsg.ID), "{1:}{2:} control message of cmd{:3} and size{:4} failed decoding at field{:5}.")
+	i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrUnknownMsg.ID), "{1:}{2:} unknown message type{:3}.")
+	i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrUnknownControlMsg.ID), "{1:}{2:} unknown control command{:3}.")
+}
+
+// NewErrInvalidMsg returns an error with the ErrInvalidMsg ID.
+func NewErrInvalidMsg(ctx *context.T, typ byte, size int64, field int64) error {
+	return verror.New(ErrInvalidMsg, ctx, typ, size, field)
+}
+
+// NewErrInvalidControlMsg returns an error with the ErrInvalidControlMsg ID.
+func NewErrInvalidControlMsg(ctx *context.T, cmd byte, size int64, field int64) error {
+	return verror.New(ErrInvalidControlMsg, ctx, cmd, size, field)
+}
+
+// NewErrUnknownMsg returns an error with the ErrUnknownMsg ID.
+func NewErrUnknownMsg(ctx *context.T, typ byte) error {
+	return verror.New(ErrUnknownMsg, ctx, typ)
+}
+
+// NewErrUnknownControlMsg returns an error with the ErrUnknownControlMsg ID.
+func NewErrUnknownControlMsg(ctx *context.T, cmd byte) error {
+	return verror.New(ErrUnknownControlMsg, ctx, cmd)
+}
diff --git a/runtime/internal/flow/conn/message.go b/runtime/internal/flow/conn/message.go
new file mode 100644
index 0000000..7e9f335
--- /dev/null
+++ b/runtime/internal/flow/conn/message.go
@@ -0,0 +1,329 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package conn
+
+import (
+	"errors"
+
+	"v.io/v23/context"
+	"v.io/v23/flow"
+	"v.io/v23/naming"
+	"v.io/v23/rpc/version"
+)
+
+// TODO(mattr): Link to protocol doc.
+
+type message interface {
+	write(ctx *context.T, p *messagePipe) error
+	read(ctx *context.T, data []byte) error
+}
+
+// message types.
+const (
+	invalidType = iota
+	controlType
+	dataType
+	unencryptedDataType
+)
+
+// control commands.
+const (
+	invalidCmd = iota
+	setupCmd
+	tearDownCmd
+	openFlowCmd
+	addRecieveBuffersCmd
+)
+
+// setup options.
+const (
+	invalidOption = iota
+	peerNaClPublicKeyOption
+	peerRemoteEndpointOption
+	peerLocalEndpointOption
+)
+
+// data flags.
+const (
+	closeFlag = 1 << iota
+	metadataFlag
+)
+
+// random consts.
+const (
+	maxVarUint64Size = 9
+)
+
+// setup is the first message over the wire.  It negotiates protocol version
+// and encryption options for connection.
+type setup struct {
+	versions           version.RPCVersionRange
+	PeerNaClPublicKey  *[32]byte
+	PeerRemoteEndpoint naming.Endpoint
+	PeerLocalEndpoint  naming.Endpoint
+}
+
+func (m *setup) write(ctx *context.T, p *messagePipe) error {
+	p.controlBuf = writeVarUint64(uint64(m.versions.Min), p.controlBuf[:0])
+	p.controlBuf = writeVarUint64(uint64(m.versions.Max), p.controlBuf)
+	return p.write([][]byte{{controlType}}, [][]byte{{setupCmd}, p.controlBuf})
+}
+func (m *setup) read(ctx *context.T, orig []byte) error {
+	var (
+		data  = orig
+		valid bool
+		v     uint64
+	)
+	if v, data, valid = readVarUint64(ctx, data); !valid {
+		return NewErrInvalidControlMsg(ctx, setupCmd, int64(len(orig)), 0)
+	}
+	m.versions.Min = version.RPCVersion(v)
+	if v, data, valid = readVarUint64(ctx, data); !valid {
+		return NewErrInvalidControlMsg(ctx, setupCmd, int64(len(orig)), 1)
+	}
+	m.versions.Max = version.RPCVersion(v)
+	return nil
+}
+
+// tearDown is sent over the wire before a connection is closed.
+type tearDown struct {
+	Err error
+}
+
+func (m *tearDown) write(ctx *context.T, p *messagePipe) error {
+	return p.write([][]byte{{controlType}}, [][]byte{{tearDownCmd}, []byte(m.Err.Error())})
+}
+func (m *tearDown) read(ctx *context.T, data []byte) error {
+	m.Err = errors.New(string(data))
+	return nil
+}
+
+// openFlow is sent at the beginning of every new flow.
+type openFlow struct {
+	id              flowID
+	initialCounters uint64
+}
+
+func (m *openFlow) write(ctx *context.T, p *messagePipe) error {
+	p.controlBuf = writeVarUint64(uint64(m.id), p.controlBuf[:0])
+	p.controlBuf = writeVarUint64(m.initialCounters, p.controlBuf)
+	return p.write([][]byte{{controlType}}, [][]byte{{openFlowCmd}, p.controlBuf})
+}
+func (m *openFlow) read(ctx *context.T, orig []byte) error {
+	var (
+		data  = orig
+		valid bool
+		v     uint64
+	)
+	if v, data, valid = readVarUint64(ctx, data); !valid {
+		return NewErrInvalidControlMsg(ctx, openFlowCmd, int64(len(orig)), 0)
+	}
+	m.id = flowID(v)
+	if m.initialCounters, data, valid = readVarUint64(ctx, data); !valid {
+		return NewErrInvalidControlMsg(ctx, openFlowCmd, int64(len(orig)), 1)
+	}
+	return nil
+}
+
+// addRecieveBuffers is sent as flows are read from locally.  The counters
+// inform remote writers that there is local buffer space available.
+type addRecieveBuffers struct {
+	counters map[flowID]uint64
+}
+
+func (m *addRecieveBuffers) write(ctx *context.T, p *messagePipe) error {
+	p.controlBuf = p.controlBuf[:0]
+	for fid, val := range m.counters {
+		p.controlBuf = writeVarUint64(uint64(fid), p.controlBuf)
+		p.controlBuf = writeVarUint64(val, p.controlBuf)
+	}
+	return p.write([][]byte{{controlType}}, [][]byte{{addRecieveBuffersCmd}, p.controlBuf})
+}
+func (m *addRecieveBuffers) read(ctx *context.T, orig []byte) error {
+	var (
+		data     = orig
+		valid    bool
+		fid, val uint64
+		n        int64
+	)
+	m.counters = map[flowID]uint64{}
+	for len(data) > 0 {
+		if fid, data, valid = readVarUint64(ctx, data); !valid {
+			return NewErrInvalidControlMsg(ctx, addRecieveBuffersCmd, int64(len(orig)), n)
+		}
+		if val, data, valid = readVarUint64(ctx, data); !valid {
+			return NewErrInvalidControlMsg(ctx, addRecieveBuffersCmd, int64(len(orig)), n+1)
+		}
+		m.counters[flowID(fid)] = val
+		n += 2
+	}
+	return nil
+}
+
+// data carries encrypted data for a specific flow.
+type data struct {
+	id      flowID
+	flags   uint64
+	payload []byte
+}
+
+func (m *data) write(ctx *context.T, p *messagePipe) error {
+	p.dataBuf = writeVarUint64(uint64(m.id), p.dataBuf[:0])
+	p.dataBuf = writeVarUint64(m.flags, p.dataBuf)
+	return p.write([][]byte{{dataType}}, [][]byte{p.dataBuf, m.payload})
+}
+func (m *data) read(ctx *context.T, orig []byte) error {
+	var (
+		data  = orig
+		valid bool
+		v     uint64
+	)
+	if v, data, valid = readVarUint64(ctx, data); !valid {
+		return NewErrInvalidMsg(ctx, dataType, int64(len(orig)), 0)
+	}
+	m.id = flowID(v)
+	if m.flags, data, valid = readVarUint64(ctx, data); !valid {
+		return NewErrInvalidMsg(ctx, dataType, int64(len(orig)), 1)
+	}
+	m.payload = data
+	return nil
+}
+
+// unencryptedData carries unencrypted data for a specific flow.
+type unencryptedData struct {
+	id      flowID
+	flags   uint64
+	payload []byte
+}
+
+func (m *unencryptedData) write(ctx *context.T, p *messagePipe) error {
+	p.dataBuf = writeVarUint64(uint64(m.id), p.dataBuf[:0])
+	p.dataBuf = writeVarUint64(m.flags, p.dataBuf)
+	// re-use the controlBuf for the data size.
+	p.controlBuf = writeVarUint64(uint64(len(m.payload)), p.controlBuf[:0])
+	return p.write(
+		[][]byte{{unencryptedDataType}, p.controlBuf, m.payload},
+		[][]byte{p.dataBuf})
+}
+func (m *unencryptedData) read(ctx *context.T, orig []byte) error {
+	var (
+		data  = orig
+		valid bool
+		v     uint64
+	)
+	if v, data, valid = readVarUint64(ctx, data); !valid {
+		return NewErrInvalidMsg(ctx, unencryptedDataType, int64(len(orig)), 0)
+	}
+	plen := int(v)
+	if plen > len(data) {
+		return NewErrInvalidMsg(ctx, unencryptedDataType, int64(len(orig)), 1)
+	}
+	m.payload, data = data[:plen], data[plen:]
+	if v, data, valid = readVarUint64(ctx, data); !valid {
+		return NewErrInvalidMsg(ctx, unencryptedDataType, int64(len(orig)), 2)
+	}
+	m.id = flowID(v)
+	if m.flags, data, valid = readVarUint64(ctx, data); !valid {
+		return NewErrInvalidMsg(ctx, unencryptedDataType, int64(len(orig)), 3)
+	}
+	return nil
+}
+
+type messagePipe struct {
+	rw         flow.MsgReadWriter
+	controlBuf []byte
+	dataBuf    []byte
+	outBuf     [][]byte
+}
+
+func newMessagePipe(rw flow.MsgReadWriter) *messagePipe {
+	return &messagePipe{
+		rw:         rw,
+		controlBuf: make([]byte, 256),
+		dataBuf:    make([]byte, 2*maxVarUint64Size),
+		outBuf:     make([][]byte, 5),
+	}
+}
+
+func (p *messagePipe) write(unencrypted [][]byte, encrypted [][]byte) error {
+	p.outBuf = append(p.outBuf[:0], unencrypted...)
+	p.outBuf = append(p.outBuf, encrypted...)
+	_, err := p.rw.WriteMsg(p.outBuf...)
+	return err
+}
+
+func (p *messagePipe) writeMsg(ctx *context.T, m message) error {
+	return m.write(ctx, p)
+}
+
+func (p *messagePipe) readMsg(ctx *context.T) (message, error) {
+	msg, err := p.rw.ReadMsg()
+	if len(msg) == 0 {
+		return nil, NewErrInvalidMsg(ctx, invalidType, 0, 0)
+	}
+	if err != nil {
+		return nil, err
+	}
+	var m message
+	switch msg[0] {
+	case controlType:
+		if len(msg) == 1 {
+			return nil, NewErrInvalidControlMsg(ctx, invalidCmd, 0, 1)
+		}
+		msg = msg[1:]
+		switch msg[0] {
+		case setupCmd:
+			m = &setup{}
+		case tearDownCmd:
+			m = &tearDown{}
+		case openFlowCmd:
+			m = &openFlow{}
+		case addRecieveBuffersCmd:
+			m = &addRecieveBuffers{}
+		default:
+			return nil, NewErrUnknownControlMsg(ctx, msg[0])
+		}
+	case dataType:
+		m = &data{}
+	case unencryptedDataType:
+		m = &unencryptedData{}
+	default:
+		return nil, NewErrUnknownMsg(ctx, msg[0])
+	}
+	return m, m.read(ctx, msg[1:])
+}
+
+func readVarUint64(ctx *context.T, data []byte) (uint64, []byte, bool) {
+	if len(data) == 0 {
+		return 0, data, false
+	}
+	l := data[0]
+	if l <= 0x7f {
+		return uint64(l), data[1:], true
+	}
+	l = 0xff - l + 1
+	if l > 8 || len(data)-1 < int(l) {
+		return 0, data, false
+	}
+	var out uint64
+	for i := 1; i < int(l+1); i++ {
+		out = out<<8 | uint64(data[i])
+	}
+	return out, data[l+1:], true
+}
+
+func writeVarUint64(u uint64, buf []byte) []byte {
+	if u <= 0x7f {
+		return append(buf, byte(u))
+	}
+	shift, l := 56, byte(7)
+	for ; shift >= 0 && (u>>uint(shift))&0xff == 0; shift, l = shift-8, l-1 {
+	}
+	buf = append(buf, 0xff-l)
+	for ; shift >= 0; shift -= 8 {
+		buf = append(buf, byte(u>>uint(shift))&0xff)
+	}
+	return buf
+}
diff --git a/runtime/internal/flow/conn/message_test.go b/runtime/internal/flow/conn/message_test.go
new file mode 100644
index 0000000..782a5a2
--- /dev/null
+++ b/runtime/internal/flow/conn/message_test.go
@@ -0,0 +1,122 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package conn
+
+import (
+	"errors"
+	"reflect"
+	"testing"
+
+	"v.io/v23/context"
+	"v.io/v23/rpc/version"
+)
+
+func TestVarInt(t *testing.T) {
+	cases := []uint64{
+		0x00, 0x01,
+		0x7f, 0x80,
+		0xff, 0x100,
+		0xffff, 0x10000,
+		0xffffff, 0x1000000,
+		0xffffffff, 0x100000000,
+		0xffffffffff, 0x10000000000,
+		0xffffffffffff, 0x1000000000000,
+		0xffffffffffffff, 0x100000000000000,
+		0xffffffffffffffff,
+	}
+	ctx, cancel := context.RootContext()
+	defer cancel()
+	for _, want := range cases {
+		got, b, valid := readVarUint64(ctx, writeVarUint64(want, []byte{}))
+		if !valid {
+			t.Fatalf("error reading %x", want)
+		}
+		if len(b) != 0 {
+			t.Errorf("unexpected buffer remaining for %x: %v", want, b)
+		}
+		if got != want {
+			t.Errorf("got: %d want: %d", got, want)
+		}
+	}
+}
+
+type fakeMsgReadWriter struct {
+	bufs [][]byte
+	t    *testing.T
+}
+
+func (f *fakeMsgReadWriter) WriteMsg(data ...[]byte) (int, error) {
+	buf := []byte{}
+	for _, d := range data {
+		buf = append(buf, d...)
+	}
+	f.bufs = append(f.bufs, buf)
+	return len(buf), nil
+}
+
+func (f *fakeMsgReadWriter) ReadMsg() (buf []byte, err error) {
+	// TODO(mattr): block if empty.
+	buf, f.bufs = f.bufs[0], f.bufs[1:]
+	f.t.Logf("reading: %v", buf)
+	return buf, nil
+}
+
+func testMessages(t *testing.T, cases []message) {
+	ctx, cancel := context.RootContext()
+	defer cancel()
+	p := newMessagePipe(&fakeMsgReadWriter{t: t})
+	for _, want := range cases {
+		if err := p.writeMsg(ctx, want); err != nil {
+			t.Errorf("unexpected error for %#v: %v", want, err)
+		}
+		got, err := p.readMsg(ctx)
+		if err != nil {
+			t.Errorf("unexpected error reading %#v: %v", want, err)
+		}
+		if !reflect.DeepEqual(got, want) {
+			t.Errorf("got: %#v, want %#v", got, want)
+		}
+	}
+}
+
+func TestSetup(t *testing.T) {
+	testMessages(t, []message{
+		&setup{versions: version.RPCVersionRange{Min: 3, Max: 5}},
+	})
+}
+
+func TestTearDown(t *testing.T) {
+	testMessages(t, []message{
+		&tearDown{Err: errors.New("foobar")},
+	})
+}
+
+func TestOpenFlow(t *testing.T) {
+	testMessages(t, []message{
+		&openFlow{id: 23, initialCounters: 1 << 20},
+	})
+}
+
+func TestAddReceiveBuffers(t *testing.T) {
+	testMessages(t, []message{
+		&addRecieveBuffers{counters: map[flowID]uint64{}},
+		&addRecieveBuffers{counters: map[flowID]uint64{
+			4: 233,
+			9: 423242,
+		}},
+	})
+}
+
+func TestData(t *testing.T) {
+	testMessages(t, []message{
+		&data{id: 1123, flags: 232, payload: []byte("fake payload")},
+	})
+}
+
+func TestUnencryptedData(t *testing.T) {
+	testMessages(t, []message{
+		&unencryptedData{id: 1123, flags: 232, payload: []byte("fake payload")},
+	})
+}