jsonrpc2: JSON-RPC 2.0 implementation See https://www.jsonrpc.org/specification. Forked from golang.org/x/tool/internal/jsonrpc2_v2 at commit 13cf844527f48de3074270bb538dff67117a6655. Changed to use golang.org/x/exp/event. This CL also contains a fork of golang.org/x/tools/internal/stack/... at the same commit, needed for testing. Change-Id: Ifc634c2b8e368a51fc43a7c0fab46cfa0ed5464b Reviewed-on: https://go-review.googlesource.com/c/exp/+/326676 Trust: Jonathan Amsterdam <jba@google.com> Run-TryBot: Jonathan Amsterdam <jba@google.com> Reviewed-by: Ian Cottrell <iancottrell@google.com>

diff --git a/internal/stack/gostacks/gostacks.go b/internal/stack/gostacks/gostacks.go
new file mode 100644
index 0000000..b2fb4e4
--- /dev/null
+++ b/internal/stack/gostacks/gostacks.go

@@ -0,0 +1,23 @@
+// Copyright 2020 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// The gostacks command processes stdin looking for things that look like
+// stack traces and simplifying them to make the log more readable.
+// It collates stack traces that have the same path as well as simplifying the
+// individual lines of the trace.
+// The processed log is printed to stdout.
+package main
+
+import (
+	"fmt"
+	"os"
+
+	"golang.org/x/exp/internal/stack"
+)
+
+func main() {
+	if err := stack.Process(os.Stdout, os.Stdin); err != nil {
+		fmt.Fprintln(os.Stderr, err)
+	}
+}

diff --git a/internal/stack/parse.go b/internal/stack/parse.go
new file mode 100644
index 0000000..e01da8f
--- /dev/null
+++ b/internal/stack/parse.go

@@ -0,0 +1,175 @@
+// Copyright 2020 The Go 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 stack
+
+import (
+	"bufio"
+	"errors"
+	"io"
+	"regexp"
+	"strconv"
+)
+
+var (
+	reBlank     = regexp.MustCompile(`^\s*$`)
+	reGoroutine = regexp.MustCompile(`^\s*goroutine (\d+) \[([^\]]*)\]:\s*$`)
+	reCall      = regexp.MustCompile(`^\s*` +
+		`(created by )?` + //marker
+		`(([\w/.]+/)?[\w]+)\.` + //package
+		`(\(([^:.)]*)\)\.)?` + //optional type
+		`([\w\.]+)` + //function
+		`(\(.*\))?` + // args
+		`\s*$`)
+	rePos = regexp.MustCompile(`^\s*(.*):(\d+)( .*)?$`)
+
+	errBreakParse = errors.New("break parse")
+)
+
+// Scanner splits an input stream into lines in a way that is consumable by
+// the parser.
+type Scanner struct {
+	lines *bufio.Scanner
+	done  bool
+}
+
+// NewScanner creates a scanner on top of a reader.
+func NewScanner(r io.Reader) *Scanner {
+	s := &Scanner{
+		lines: bufio.NewScanner(r),
+	}
+	s.Skip() // prefill
+	return s
+}
+
+// Peek returns the next line without consuming it.
+func (s *Scanner) Peek() string {
+	if s.done {
+		return ""
+	}
+	return s.lines.Text()
+}
+
+// Skip consumes the next line without looking at it.
+// Normally used after it has already been looked at using Peek.
+func (s *Scanner) Skip() {
+	if !s.lines.Scan() {
+		s.done = true
+	}
+}
+
+// Next consumes and returns the next line.
+func (s *Scanner) Next() string {
+	line := s.Peek()
+	s.Skip()
+	return line
+}
+
+// Done returns true if the scanner has reached the end of the underlying
+// stream.
+func (s *Scanner) Done() bool {
+	return s.done
+}
+
+// Err returns true if the scanner has reached the end of the underlying
+// stream.
+func (s *Scanner) Err() error {
+	return s.lines.Err()
+}
+
+// Match returns the submatchs of the regular expression against the next line.
+// If it matched the line is also consumed.
+func (s *Scanner) Match(re *regexp.Regexp) []string {
+	if s.done {
+		return nil
+	}
+	match := re.FindStringSubmatch(s.Peek())
+	if match != nil {
+		s.Skip()
+	}
+	return match
+}
+
+// SkipBlank skips any number of pure whitespace lines.
+func (s *Scanner) SkipBlank() {
+	for !s.done {
+		line := s.Peek()
+		if len(line) != 0 && !reBlank.MatchString(line) {
+			return
+		}
+		s.Skip()
+	}
+}
+
+// Parse the current contiguous block of goroutine stack traces until the
+// scanned content no longer matches.
+func Parse(scanner *Scanner) (Dump, error) {
+	dump := Dump{}
+	for {
+		gr, ok := parseGoroutine(scanner)
+		if !ok {
+			return dump, nil
+		}
+		dump = append(dump, gr)
+	}
+}
+
+func parseGoroutine(scanner *Scanner) (Goroutine, bool) {
+	match := scanner.Match(reGoroutine)
+	if match == nil {
+		return Goroutine{}, false
+	}
+	id, _ := strconv.ParseInt(match[1], 0, 32)
+	gr := Goroutine{
+		ID:    int(id),
+		State: match[2],
+	}
+	for {
+		frame, ok := parseFrame(scanner)
+		if !ok {
+			scanner.SkipBlank()
+			return gr, true
+		}
+		if frame.Position.Filename != "" {
+			gr.Stack = append(gr.Stack, frame)
+		}
+	}
+}
+
+func parseFrame(scanner *Scanner) (Frame, bool) {
+	fun, ok := parseFunction(scanner)
+	if !ok {
+		return Frame{}, false
+	}
+	frame := Frame{
+		Function: fun,
+	}
+	frame.Position, ok = parsePosition(scanner)
+	// if ok is false, then this is a broken state.
+	// we got the func but not the file that must follow
+	// the consumed line can be recovered from the frame
+	//TODO: push back the fun raw
+	return frame, ok
+}
+
+func parseFunction(scanner *Scanner) (Function, bool) {
+	match := scanner.Match(reCall)
+	if match == nil {
+		return Function{}, false
+	}
+	return Function{
+		Package: match[2],
+		Type:    match[5],
+		Name:    match[6],
+	}, true
+}
+
+func parsePosition(scanner *Scanner) (Position, bool) {
+	match := scanner.Match(rePos)
+	if match == nil {
+		return Position{}, false
+	}
+	line, _ := strconv.ParseInt(match[2], 0, 32)
+	return Position{Filename: match[1], Line: int(line)}, true
+}

diff --git a/internal/stack/process.go b/internal/stack/process.go
new file mode 100644
index 0000000..ac19366
--- /dev/null
+++ b/internal/stack/process.go

@@ -0,0 +1,112 @@
+// Copyright 2020 The Go 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 stack
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"runtime"
+	"sort"
+)
+
+// Capture get the current stack traces from the runtime.
+func Capture() Dump {
+	buf := make([]byte, 2<<20)
+	buf = buf[:runtime.Stack(buf, true)]
+	scanner := NewScanner(bytes.NewReader(buf))
+	dump, _ := Parse(scanner)
+	return dump
+}
+
+// Summarize a dump for easier consumption.
+// This collates goroutines with equivalent stacks.
+func Summarize(dump Dump) Summary {
+	s := Summary{
+		Total: len(dump),
+	}
+	for _, gr := range dump {
+		s.addGoroutine(gr)
+	}
+	return s
+}
+
+// Process and input stream to an output stream, summarizing any stacks that
+// are detected in place.
+func Process(out io.Writer, in io.Reader) error {
+	scanner := NewScanner(in)
+	for {
+		dump, err := Parse(scanner)
+		summary := Summarize(dump)
+		switch {
+		case len(dump) > 0:
+			fmt.Fprintf(out, "%+v\n\n", summary)
+		case err != nil:
+			return err
+		case scanner.Done():
+			return scanner.Err()
+		default:
+			// must have been a line that is not part of a dump
+			fmt.Fprintln(out, scanner.Next())
+		}
+	}
+}
+
+// Diff calculates the delta between two dumps.
+func Diff(before, after Dump) Delta {
+	result := Delta{}
+	processed := make(map[int]bool)
+	for _, gr := range before {
+		processed[gr.ID] = false
+	}
+	for _, gr := range after {
+		if _, found := processed[gr.ID]; found {
+			result.Shared = append(result.Shared, gr)
+		} else {
+			result.After = append(result.After, gr)
+		}
+		processed[gr.ID] = true
+	}
+	for _, gr := range before {
+		if done := processed[gr.ID]; !done {
+			result.Before = append(result.Before, gr)
+		}
+	}
+	return result
+}
+
+// TODO: do we want to allow contraction of stacks before comparison?
+func (s *Summary) addGoroutine(gr Goroutine) {
+	index := sort.Search(len(s.Calls), func(i int) bool {
+		return !s.Calls[i].Stack.less(gr.Stack)
+	})
+	if index >= len(s.Calls) || !s.Calls[index].Stack.equal(gr.Stack) {
+		// insert new stack, first increase the length
+		s.Calls = append(s.Calls, Call{})
+		// move the top part upward to make space
+		copy(s.Calls[index+1:], s.Calls[index:])
+		// insert the new call
+		s.Calls[index] = Call{
+			Stack: gr.Stack,
+		}
+	}
+	// merge the goroutine into the matched call
+	s.Calls[index].merge(gr)
+}
+
+//TODO: do we want other grouping strategies?
+func (c *Call) merge(gr Goroutine) {
+	for i := range c.Groups {
+		canditate := &c.Groups[i]
+		if canditate.State == gr.State {
+			canditate.Goroutines = append(canditate.Goroutines, gr)
+			return
+		}
+	}
+	c.Groups = append(c.Groups, Group{
+		State:      gr.State,
+		Goroutines: []Goroutine{gr},
+	})
+}

diff --git a/internal/stack/stack.go b/internal/stack/stack.go
new file mode 100644
index 0000000..479301a
--- /dev/null
+++ b/internal/stack/stack.go

@@ -0,0 +1,170 @@
+// Copyright 2020 The Go 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 stack provides support for parsing standard goroutine stack traces.
+package stack
+
+import (
+	"fmt"
+	"text/tabwriter"
+)
+
+// Dump is a raw set of goroutines and their stacks.
+type Dump []Goroutine
+
+// Goroutine is a single parsed goroutine dump.
+type Goroutine struct {
+	State string // state that the goroutine is in.
+	ID    int    // id of the goroutine.
+	Stack Stack  // call frames that make up the stack
+}
+
+// Stack is a set of frames in a callstack.
+type Stack []Frame
+
+// Frame is a point in a call stack.
+type Frame struct {
+	Function Function
+	Position Position
+}
+
+// Function is the function called at a frame.
+type Function struct {
+	Package string // package name of function if known
+	Type    string // if set function is a method of this type
+	Name    string // function name of the frame
+}
+
+// Position is the file position for a frame.
+type Position struct {
+	Filename string // source filename
+	Line     int    // line number within file
+}
+
+// Summary is a set of stacks processed and collated into Calls.
+type Summary struct {
+	Total int    // the total count of goroutines in the summary
+	Calls []Call // the collated stack traces
+}
+
+// Call is set of goroutines that all share the same callstack.
+// They will be grouped by state.
+type Call struct {
+	Stack  Stack   // the shared callstack information
+	Groups []Group // the sets of goroutines with the same state
+}
+
+// Group is a set of goroutines with the same stack that are in the same state.
+type Group struct {
+	State      string      // the shared state of the goroutines
+	Goroutines []Goroutine // the set of goroutines in this group
+}
+
+// Delta represents the difference between two stack dumps.
+type Delta struct {
+	Before Dump // The goroutines that were only in the before set.
+	Shared Dump // The goroutines that were in both sets.
+	After  Dump // The goroutines that were only in the after set.
+}
+
+func (s Stack) equal(other Stack) bool {
+	if len(s) != len(other) {
+		return false
+	}
+	for i, frame := range s {
+		if !frame.equal(other[i]) {
+			return false
+		}
+	}
+	return true
+}
+
+func (s Stack) less(other Stack) bool {
+	for i, frame := range s {
+		if i >= len(other) {
+			return false
+		}
+		if frame.less(other[i]) {
+			return true
+		}
+		if !frame.equal(other[i]) {
+			return false
+		}
+	}
+	return len(s) < len(other)
+}
+
+func (f Frame) equal(other Frame) bool {
+	return f.Position.equal(other.Position)
+}
+
+func (f Frame) less(other Frame) bool {
+	return f.Position.less(other.Position)
+}
+
+func (p Position) equal(other Position) bool {
+	return p.Filename == other.Filename && p.Line == other.Line
+}
+
+func (p Position) less(other Position) bool {
+	if p.Filename < other.Filename {
+		return true
+	}
+	if p.Filename > other.Filename {
+		return false
+	}
+	return p.Line < other.Line
+}
+
+func (s Summary) Format(w fmt.State, r rune) {
+	tw := tabwriter.NewWriter(w, 0, 0, 1, ' ', 0)
+	for i, c := range s.Calls {
+		if i > 0 {
+			fmt.Fprintf(tw, "\n\n")
+			tw.Flush()
+		}
+		fmt.Fprint(tw, c)
+	}
+	tw.Flush()
+	if s.Total > 0 && w.Flag('+') {
+		fmt.Fprintf(w, "\n\n%d goroutines, %d unique", s.Total, len(s.Calls))
+	}
+}
+
+func (c Call) Format(w fmt.State, r rune) {
+	for i, g := range c.Groups {
+		if i > 0 {
+			fmt.Fprint(w, " ")
+		}
+		fmt.Fprint(w, g)
+	}
+	for _, f := range c.Stack {
+		fmt.Fprintf(w, "\n%v", f)
+	}
+}
+
+func (g Group) Format(w fmt.State, r rune) {
+	fmt.Fprintf(w, "[%v]: ", g.State)
+	for i, gr := range g.Goroutines {
+		if i > 0 {
+			fmt.Fprint(w, ", ")
+		}
+		fmt.Fprintf(w, "$%d", gr.ID)
+	}
+}
+
+func (f Frame) Format(w fmt.State, c rune) {
+	fmt.Fprintf(w, "%v:\t%v", f.Position, f.Function)
+}
+
+func (f Function) Format(w fmt.State, c rune) {
+	if f.Type != "" {
+		fmt.Fprintf(w, "(%v).", f.Type)
+	}
+	fmt.Fprintf(w, "%v", f.Name)
+}
+
+func (p Position) Format(w fmt.State, c rune) {
+	fmt.Fprintf(w, "%v:%v", p.Filename, p.Line)
+}

diff --git a/internal/stack/stack_test.go b/internal/stack/stack_test.go
new file mode 100644
index 0000000..8f22236
--- /dev/null
+++ b/internal/stack/stack_test.go

@@ -0,0 +1,193 @@
+// Copyright 2020 The Go 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 stack_test
+
+import (
+	"bytes"
+	"strings"
+	"testing"
+
+	"golang.org/x/exp/internal/stack"
+)
+
+func TestProcess(t *testing.T) {
+	for _, test := range []struct{ name, input, expect string }{{
+		name:   `empty`,
+		input:  ``,
+		expect: ``,
+	}, {
+		name:  `no_frame`,
+		input: `goroutine 1 [running]:`,
+		expect: `
+[running]: $1
+
+1 goroutines, 1 unique
+`,
+	}, {
+		name: `one_frame`,
+		input: `
+goroutine 1 [running]:
+package.function(args)
+	file.go:10
+`,
+		expect: `
+[running]: $1
+file.go:10: function
+
+1 goroutines, 1 unique
+`,
+	}, {
+		name: `one_call`,
+		input: `
+goroutine 1 [running]:
+package1.functionA(args)
+	file1.go:10
+package2.functionB(args)
+	file2.go:20
+package3.functionC(args)
+	file3.go:30
+`,
+		expect: `
+[running]: $1
+file1.go:10: functionA
+file2.go:20: functionB
+file3.go:30: functionC
+
+1 goroutines, 1 unique
+`,
+	}, {
+		name: `two_call`,
+		input: `
+goroutine 1 [running]:
+package1.functionA(args)
+	file1.go:10
+goroutine 2 [running]:
+package2.functionB(args)
+	file2.go:20
+`,
+		expect: `
+[running]: $1
+file1.go:10: functionA
+
+[running]: $2
+file2.go:20: functionB
+
+2 goroutines, 2 unique
+`,
+	}, {
+		name: `merge_call`,
+		input: `
+goroutine 1 [running]:
+package1.functionA(args)
+	file1.go:10
+goroutine 2 [running]:
+package1.functionA(args)
+	file1.go:10
+`,
+		expect: `
+[running]: $1, $2
+file1.go:10: functionA
+
+2 goroutines, 1 unique
+`,
+	}, {
+		name: `alternating_call`,
+		input: `
+goroutine 1 [running]:
+package1.functionA(args)
+	file1.go:10
+goroutine 2 [running]:
+package2.functionB(args)
+	file2.go:20
+goroutine 3 [running]:
+package1.functionA(args)
+	file1.go:10
+goroutine 4 [running]:
+package2.functionB(args)
+	file2.go:20
+goroutine 5 [running]:
+package1.functionA(args)
+	file1.go:10
+goroutine 6 [running]:
+package2.functionB(args)
+	file2.go:20
+`,
+		expect: `
+[running]: $1, $3, $5
+file1.go:10: functionA
+
+[running]: $2, $4, $6
+file2.go:20: functionB
+
+6 goroutines, 2 unique
+`,
+	}, {
+		name: `sort_calls`,
+		input: `
+goroutine 1 [running]:
+package3.functionC(args)
+	file3.go:30
+goroutine 2 [running]:
+package2.functionB(args)
+	file2.go:20
+goroutine 3 [running]:
+package1.functionA(args)
+	file1.go:10
+`,
+		expect: `
+[running]: $3
+file1.go:10: functionA
+
+[running]: $2
+file2.go:20: functionB
+
+[running]: $1
+file3.go:30: functionC
+
+3 goroutines, 3 unique
+`,
+	}, {
+		name: `real_single`,
+		input: `
+panic: oops
+
+goroutine 53 [running]:
+golang.org/x/tools/internal/jsonrpc2_test.testHandler.func1(0x1240c20, 0xc000013350, 0xc0000133b0, 0x1240ca0, 0xc00002ab00, 0x3, 0x3)
+	/work/tools/internal/jsonrpc2/jsonrpc2_test.go:160 +0x74c
+golang.org/x/tools/internal/jsonrpc2.(*Conn).Run(0xc000204330, 0x1240c20, 0xc000204270, 0x1209570, 0xc000212120, 0x1242700)
+	/work/tools/internal/jsonrpc2/jsonrpc2.go:187 +0x777
+golang.org/x/tools/internal/jsonrpc2_test.run.func1(0x123ebe0, 0xc000206018, 0x123ec20, 0xc000206010, 0xc0002080a0, 0xc000204330, 0x1240c20, 0xc000204270, 0xc000212120)
+	/work/tools/internal/jsonrpc2/jsonrpc2_test.go:131 +0xe2
+created by golang.org/x/tools/internal/jsonrpc2_test.run
+	/work/tools/internal/jsonrpc2/jsonrpc2_test.go:121 +0x263
+FAIL    golang.org/x/tools/internal/jsonrpc2    0.252s
+FAIL
+`,
+		expect: `
+panic: oops
+
+[running]: $53
+/work/tools/internal/jsonrpc2/jsonrpc2_test.go:160: testHandler.func1
+/work/tools/internal/jsonrpc2/jsonrpc2.go:187:      (*Conn).Run
+/work/tools/internal/jsonrpc2/jsonrpc2_test.go:131: run.func1
+/work/tools/internal/jsonrpc2/jsonrpc2_test.go:121: run
+
+1 goroutines, 1 unique
+
+FAIL    golang.org/x/tools/internal/jsonrpc2    0.252s
+FAIL
+`,
+	}} {
+		t.Run(test.name, func(t *testing.T) {
+			buf := &bytes.Buffer{}
+			stack.Process(buf, strings.NewReader(test.input))
+			expect := strings.TrimSpace(test.expect)
+			got := strings.TrimSpace(buf.String())
+			if got != expect {
+				t.Errorf("got:\n%s\nexpect:\n%s", got, expect)
+			}
+		})
+	}
+}

diff --git a/internal/stack/stacktest/stacktest.go b/internal/stack/stacktest/stacktest.go
new file mode 100644
index 0000000..64bc6c5
--- /dev/null
+++ b/internal/stack/stacktest/stacktest.go

@@ -0,0 +1,50 @@
+// Copyright 2018 The Go 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 stacktest
+
+import (
+	"testing"
+	"time"
+
+	"golang.org/x/exp/internal/stack"
+)
+
+//this is only needed to support pre 1.14 when testing.TB did not have Cleanup
+type withCleanup interface {
+	Cleanup(func())
+}
+
+// the maximum amount of time to wait for goroutines to clean themselves up.
+const maxWait = time.Second
+
+// NoLeak checks that a test (or benchmark) does not leak any goroutines.
+func NoLeak(t testing.TB) {
+	c, ok := t.(withCleanup)
+	if !ok {
+		return
+	}
+	before := stack.Capture()
+	c.Cleanup(func() {
+		var delta stack.Delta
+		start := time.Now()
+		delay := time.Millisecond
+		for {
+			after := stack.Capture()
+			delta = stack.Diff(before, after)
+			if len(delta.After) == 0 {
+				// no leaks
+				return
+			}
+			if time.Since(start) > maxWait {
+				break
+			}
+			time.Sleep(delay)
+			delay *= 2
+		}
+		// it's been long enough, and leaks are still present
+		summary := stack.Summarize(delta.After)
+		t.Errorf("goroutine leak detected:\n%+v", summary)
+	})
+}

diff --git a/jsonrpc2/conn.go b/jsonrpc2/conn.go
new file mode 100644
index 0000000..cbc1558
--- /dev/null
+++ b/jsonrpc2/conn.go

@@ -0,0 +1,474 @@
+// Copyright 2018 The Go 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 jsonrpc2
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"sync/atomic"
+
+	"golang.org/x/exp/event"
+	errors "golang.org/x/xerrors"
+)
+
+// Binder builds a connection configuration.
+// This may be used in servers to generate a new configuration per connection.
+// ConnectionOptions itself implements Binder returning itself unmodified, to
+// allow for the simple cases where no per connection information is needed.
+type Binder interface {
+	// Bind is invoked when creating a new connection.
+	// The connection is not ready to use when Bind is called.
+	Bind(context.Context, *Connection) (ConnectionOptions, error)
+}
+
+// ConnectionOptions holds the options for new connections.
+type ConnectionOptions struct {
+	// Framer allows control over the message framing and encoding.
+	// If nil, HeaderFramer will be used.
+	Framer Framer
+	// Preempter allows registration of a pre-queue message handler.
+	// If nil, no messages will be preempted.
+	Preempter Preempter
+	// Handler is used as the queued message handler for inbound messages.
+	// If nil, all responses will be ErrNotHandled.
+	Handler Handler
+}
+
+// Connection manages the jsonrpc2 protocol, connecting responses back to their
+// calls.
+// Connection is bidirectional; it does not have a designated server or client
+// end.
+type Connection struct {
+	seq         int64 // must only be accessed using atomic operations
+	closer      io.Closer
+	writerBox   chan Writer
+	outgoingBox chan map[ID]chan<- *Response
+	incomingBox chan map[ID]*incoming
+	async       async
+}
+
+type AsyncCall struct {
+	id         ID
+	response   chan *Response // the channel a response will be delivered on
+	resultBox  chan asyncResult
+	endBuilder event.Builder // close the tracing span when all processing for the message is complete
+}
+
+type asyncResult struct {
+	result []byte
+	err    error
+}
+
+// incoming is used to track an incoming request as it is being handled
+type incoming struct {
+	request    *Request        // the request being processed
+	baseCtx    context.Context // a base context for the message processing
+	endBuilder event.Builder   // builder's End method called when all processing for the message is complete
+	handleCtx  context.Context // the context for handling the message, child of baseCtx
+	cancel     func()          // a function that cancels the handling context
+}
+
+// Bind returns the options unmodified.
+func (o ConnectionOptions) Bind(context.Context, *Connection) (ConnectionOptions, error) {
+	return o, nil
+}
+
+// newConnection creates a new connection and runs it.
+// This is used by the Dial and Serve functions to build the actual connection.
+func newConnection(ctx context.Context, rwc io.ReadWriteCloser, binder Binder) (*Connection, error) {
+	c := &Connection{
+		closer:      rwc,
+		writerBox:   make(chan Writer, 1),
+		outgoingBox: make(chan map[ID]chan<- *Response, 1),
+		incomingBox: make(chan map[ID]*incoming, 1),
+	}
+
+	options, err := binder.Bind(ctx, c)
+	if err != nil {
+		return nil, err
+	}
+	if options.Framer == nil {
+		options.Framer = HeaderFramer()
+	}
+	if options.Preempter == nil {
+		options.Preempter = defaultHandler{}
+	}
+	if options.Handler == nil {
+		options.Handler = defaultHandler{}
+	}
+	c.outgoingBox <- make(map[ID]chan<- *Response)
+	c.incomingBox <- make(map[ID]*incoming)
+	c.async.init()
+	// the goroutines started here will continue until the underlying stream is closed
+	reader := options.Framer.Reader(rwc)
+	readToQueue := make(chan *incoming)
+	queueToDeliver := make(chan *incoming)
+	go c.readIncoming(ctx, reader, readToQueue)
+	go c.manageQueue(ctx, options.Preempter, readToQueue, queueToDeliver)
+	go c.deliverMessages(ctx, options.Handler, queueToDeliver)
+	// releaseing the writer must be the last thing we do in case any requests
+	// are blocked waiting for the connection to be ready
+	c.writerBox <- options.Framer.Writer(rwc)
+	return c, nil
+}
+
+// Notify invokes the target method but does not wait for a response.
+// The params will be marshaled to JSON before sending over the wire, and will
+// be handed to the method invoked.
+func (c *Connection) Notify(ctx context.Context, method string, params interface{}) error {
+	notify, err := NewNotification(method, params)
+	if err != nil {
+		return errors.Errorf("marshaling notify parameters: %v", err)
+	}
+	b := event.To(ctx).With(Method.Of(method))
+	b.Clone().Metric(Started.Record(1))
+	ctx, endBuilder := b.With(RPCDirection.Of(Outbound)).Start(method)
+	err = c.write(ctx, notify)
+	event.To(ctx).With(Error.Of(err)).Metric(Finished.Record(1))
+	endBuilder.End()
+	return err
+}
+
+// Call invokes the target method and returns an object that can be used to await the response.
+// The params will be marshaled to JSON before sending over the wire, and will
+// be handed to the method invoked.
+// You do not have to wait for the response, it can just be ignored if not needed.
+// If sending the call failed, the response will be ready and have the error in it.
+func (c *Connection) Call(ctx context.Context, method string, params interface{}) *AsyncCall {
+	result := &AsyncCall{
+		id:        Int64ID(atomic.AddInt64(&c.seq, 1)),
+		resultBox: make(chan asyncResult, 1),
+	}
+	// generate a new request identifier
+	call, err := NewCall(result.id, method, params)
+	if err != nil {
+		//set the result to failed
+		result.resultBox <- asyncResult{err: errors.Errorf("marshaling call parameters: %w", err)}
+		return result
+	}
+	b := event.To(ctx).With(Method.Of(method))
+	b.Clone().Metric(Started.Record(1))
+	ctx, endBuilder := b.Clone().With(RPCDirection.Of(Outbound)).With(RPCID.Of(fmt.Sprintf("%q", result.id))).Start(method)
+	result.endBuilder = endBuilder
+	// We have to add ourselves to the pending map before we send, otherwise we
+	// are racing the response.
+	// rchan is buffered in case the response arrives without a listener.
+	result.response = make(chan *Response, 1)
+	pending := <-c.outgoingBox
+	pending[result.id] = result.response
+	c.outgoingBox <- pending
+	// now we are ready to send
+	if err := c.write(ctx, call); err != nil {
+		// sending failed, we will never get a response, so deliver a fake one
+		r, _ := NewResponse(result.id, nil, err)
+		c.incomingResponse(r)
+	}
+	return result
+}
+
+// ID used for this call.
+// This can be used to cancel the call if needed.
+func (a *AsyncCall) ID() ID { return a.id }
+
+// IsReady can be used to check if the result is already prepared.
+// This is guaranteed to return true on a result for which Await has already
+// returned, or a call that failed to send in the first place.
+func (a *AsyncCall) IsReady() bool {
+	select {
+	case r := <-a.resultBox:
+		a.resultBox <- r
+		return true
+	default:
+		return false
+	}
+}
+
+// Await the results of a Call.
+// The response will be unmarshaled from JSON into the result.
+func (a *AsyncCall) Await(ctx context.Context, result interface{}) error {
+	status := "NONE"
+	defer a.endBuilder.With(StatusCode.Of(status)).End()
+	var r asyncResult
+	select {
+	case response := <-a.response:
+		// response just arrived, prepare the result
+		switch {
+		case response.Error != nil:
+			r.err = response.Error
+			status = "ERROR"
+		default:
+			r.result = response.Result
+			status = "OK"
+		}
+	case r = <-a.resultBox:
+		// result already available
+	case <-ctx.Done():
+		status = "CANCELLED"
+		return ctx.Err()
+	}
+	// refill the box for the next caller
+	a.resultBox <- r
+	// and unpack the result
+	if r.err != nil {
+		return r.err
+	}
+	if result == nil || len(r.result) == 0 {
+		return nil
+	}
+	return json.Unmarshal(r.result, result)
+}
+
+// Respond deliverers a response to an incoming Call.
+// It is an error to not call this exactly once for any message for which a
+// handler has previously returned ErrAsyncResponse. It is also an error to
+// call this for any other message.
+func (c *Connection) Respond(id ID, result interface{}, rerr error) error {
+	pending := <-c.incomingBox
+	defer func() { c.incomingBox <- pending }()
+	entry, found := pending[id]
+	if !found {
+		return nil
+	}
+	delete(pending, id)
+	return c.respond(entry, result, rerr)
+}
+
+// Cancel is used to cancel an inbound message by ID, it does not cancel
+// outgoing messages.
+// This is only used inside a message handler that is layering a
+// cancellation protocol on top of JSON RPC 2.
+// It will not complain if the ID is not a currently active message, and it will
+// not cause any messages that have not arrived yet with that ID to be
+// cancelled.
+func (c *Connection) Cancel(id ID) {
+	pending := <-c.incomingBox
+	defer func() { c.incomingBox <- pending }()
+	if entry, found := pending[id]; found && entry.cancel != nil {
+		entry.cancel()
+		entry.cancel = nil
+	}
+}
+
+// Wait blocks until the connection is fully closed, but does not close it.
+func (c *Connection) Wait() error {
+	return c.async.wait()
+}
+
+// Close can be used to close the underlying stream, and then wait for the connection to
+// fully shut down.
+// This does not cancel in flight requests, but waits for them to gracefully complete.
+func (c *Connection) Close() error {
+	// close the underlying stream
+	if err := c.closer.Close(); err != nil && !isClosingError(err) {
+		return err
+	}
+	// and then wait for it to cause the connection to close
+	if err := c.Wait(); err != nil && !isClosingError(err) {
+		return err
+	}
+	return nil
+}
+
+// readIncoming collects inbound messages from the reader and delivers them, either responding
+// to outgoing calls or feeding requests to the queue.
+func (c *Connection) readIncoming(ctx context.Context, reader Reader, toQueue chan<- *incoming) {
+	defer close(toQueue)
+	for {
+		// get the next message
+		// no lock is needed, this is the only reader
+		msg, n, err := reader.Read(ctx)
+		if err != nil {
+			// The stream failed, we cannot continue
+			c.async.setError(err)
+			return
+		}
+		switch msg := msg.(type) {
+		case *Request:
+			entry := &incoming{
+				request: msg,
+			}
+			// add a span to the context for this request
+			b := event.To(ctx).With(Method.Of(msg.Method)).With(RPCDirection.Of(Inbound))
+			if msg.IsCall() {
+				b = b.With(RPCID.Of(fmt.Sprintf("%q", msg.ID)))
+			}
+			entry.baseCtx, entry.endBuilder = b.Start(msg.Method)
+			b = event.To(entry.baseCtx).With(Method.Of(msg.Method))
+			b.Clone().Metric(Started.Record(1))
+			b.Metric(ReceivedBytes.Record(n))
+			// in theory notifications cannot be cancelled, but we build them a cancel context anyway
+			entry.handleCtx, entry.cancel = context.WithCancel(entry.baseCtx)
+			// if the request is a call, add it to the incoming map so it can be
+			// cancelled by id
+			if msg.IsCall() {
+				pending := <-c.incomingBox
+				c.incomingBox <- pending
+				pending[msg.ID] = entry
+			}
+			// send the message to the incoming queue
+			toQueue <- entry
+		case *Response:
+			// If method is not set, this should be a response, in which case we must
+			// have an id to send the response back to the caller.
+			c.incomingResponse(msg)
+		}
+	}
+}
+
+func (c *Connection) incomingResponse(msg *Response) {
+	pending := <-c.outgoingBox
+	response, ok := pending[msg.ID]
+	if ok {
+		delete(pending, msg.ID)
+	}
+	c.outgoingBox <- pending
+	if response != nil {
+		response <- msg
+	}
+}
+
+// manageQueue reads incoming requests, attempts to proccess them with the preempter, or queue them
+// up for normal handling.
+func (c *Connection) manageQueue(ctx context.Context, preempter Preempter, fromRead <-chan *incoming, toDeliver chan<- *incoming) {
+	defer close(toDeliver)
+	q := []*incoming{}
+	ok := true
+	for {
+		var nextReq *incoming
+		if len(q) == 0 {
+			// no messages in the queue
+			// if we were closing, then we are done
+			if !ok {
+				return
+			}
+			// not closing, but nothing in the queue, so just block waiting for a read
+			nextReq, ok = <-fromRead
+		} else {
+			// we have a non empty queue, so pick whichever of reading or delivering
+			// that we can make progress on
+			select {
+			case nextReq, ok = <-fromRead:
+			case toDeliver <- q[0]:
+				//TODO: this causes a lot of shuffling, should we use a growing ring buffer? compaction?
+				q = q[1:]
+			}
+		}
+		if nextReq != nil {
+			// TODO: should we allow to limit the queue size?
+			var result interface{}
+			rerr := nextReq.handleCtx.Err()
+			if rerr == nil {
+				// only preempt if not already cancelled
+				result, rerr = preempter.Preempt(nextReq.handleCtx, nextReq.request)
+			}
+			switch {
+			case rerr == ErrNotHandled:
+				// message not handled, add it to the queue for the main handler
+				q = append(q, nextReq)
+			case rerr == ErrAsyncResponse:
+				// message handled but the response will come later
+			default:
+				// anything else means the message is fully handled
+				c.reply(nextReq, result, rerr)
+			}
+		}
+	}
+}
+
+func (c *Connection) deliverMessages(ctx context.Context, handler Handler, fromQueue <-chan *incoming) {
+	defer c.async.done()
+	for entry := range fromQueue {
+		// cancel any messages in the queue that we have a pending cancel for
+		var result interface{}
+		rerr := entry.handleCtx.Err()
+		if rerr == nil {
+			// only deliver if not already cancelled
+			result, rerr = handler.Handle(entry.handleCtx, entry.request)
+		}
+		switch {
+		case rerr == ErrNotHandled:
+			// message not handled, report it back to the caller as an error
+			c.reply(entry, nil, errors.Errorf("%w: %q", ErrMethodNotFound, entry.request.Method))
+		case rerr == ErrAsyncResponse:
+			// message handled but the response will come later
+		default:
+			c.reply(entry, result, rerr)
+		}
+	}
+}
+
+// reply is used to reply to an incoming request that has just been handled
+func (c *Connection) reply(entry *incoming, result interface{}, rerr error) {
+	if entry.request.IsCall() {
+		// we have a call finishing, remove it from the incoming map
+		pending := <-c.incomingBox
+		defer func() { c.incomingBox <- pending }()
+		delete(pending, entry.request.ID)
+	}
+	if err := c.respond(entry, result, rerr); err != nil {
+		// no way to propagate this error
+		//TODO: should we do more than just log it?
+		event.To(entry.baseCtx).With(Error.Of(err)).Log("jsonrpc2 message delivery failed")
+	}
+}
+
+// respond sends a response.
+// This is the code shared between reply and SendResponse.
+func (c *Connection) respond(entry *incoming, result interface{}, rerr error) error {
+	var err error
+	if entry.request.IsCall() {
+		// send the response
+		if result == nil && rerr == nil {
+			// call with no response, send an error anyway
+			rerr = errors.Errorf("%w: %q produced no response", ErrInternal, entry.request.Method)
+		}
+		var response *Response
+		response, err = NewResponse(entry.request.ID, result, rerr)
+		if err == nil {
+			// we write the response with the base context, in case the message was cancelled
+			err = c.write(entry.baseCtx, response)
+		}
+	} else {
+		switch {
+		case rerr != nil:
+			// notification failed
+			err = errors.Errorf("%w: %q notification failed: %v", ErrInternal, entry.request.Method, rerr)
+			rerr = nil
+		case result != nil:
+			//notification produced a response, which is an error
+			err = errors.Errorf("%w: %q produced unwanted response", ErrInternal, entry.request.Method)
+		default:
+			// normal notification finish
+		}
+	}
+	var status string
+	switch {
+	case rerr != nil || err != nil:
+		status = "ERROR"
+	default:
+		status = "OK"
+	}
+	// and just to be clean, invoke and clear the cancel if needed
+	if entry.cancel != nil {
+		entry.cancel()
+		entry.cancel = nil
+	}
+	// mark the entire request processing as done
+	entry.endBuilder.With(StatusCode.Of(status)).End()
+	return err
+}
+
+// write is used by all things that write outgoing messages, including replies.
+// it makes sure that writes are atomic
+func (c *Connection) write(ctx context.Context, msg Message) error {
+	writer := <-c.writerBox
+	defer func() { c.writerBox <- writer }()
+	n, err := writer.Write(ctx, msg)
+	// TODO: get a method label in here somehow.
+	event.To(ctx).Metric(SentBytes.Record(n))
+	return err
+}

diff --git a/jsonrpc2/defs.go b/jsonrpc2/defs.go
new file mode 100644
index 0000000..f40800d
--- /dev/null
+++ b/jsonrpc2/defs.go

@@ -0,0 +1,31 @@
+// Copyright 2021 The Go 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 jsonrpc2
+
+import (
+	"golang.org/x/exp/event"
+	"golang.org/x/exp/event/keys"
+)
+
+var (
+	Method       = keys.String("method")
+	RPCID        = keys.String("id")
+	RPCDirection = keys.String("direction")
+	Error        = keys.Value("error")
+	StatusCode   = keys.String("status.code")
+)
+
+var (
+	Started       = event.NewCounter("started", "Count of started RPCs.")
+	Finished      = event.NewCounter("finished", "Count of finished RPCs (includes error).")
+	ReceivedBytes = event.NewIntDistribution("received_bytes", "Bytes received.") //, unit.Bytes)
+	SentBytes     = event.NewIntDistribution("sent_bytes", "Bytes sent.")         //, unit.Bytes)
+	Latency       = event.NewDuration("latency", "Elapsed time of an RPC.")       //, unit.Milliseconds)
+)
+
+const (
+	Inbound  = "in"
+	Outbound = "out"
+)

diff --git a/jsonrpc2/frame.go b/jsonrpc2/frame.go
new file mode 100644
index 0000000..634717c
--- /dev/null
+++ b/jsonrpc2/frame.go

@@ -0,0 +1,179 @@
+// Copyright 2018 The Go 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 jsonrpc2
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"strconv"
+	"strings"
+
+	errors "golang.org/x/xerrors"
+)
+
+// Reader abstracts the transport mechanics from the JSON RPC protocol.
+// A Conn reads messages from the reader it was provided on construction,
+// and assumes that each call to Read fully transfers a single message,
+// or returns an error.
+// A reader is not safe for concurrent use, it is expected it will be used by
+// a single Conn in a safe manner.
+type Reader interface {
+	// Read gets the next message from the stream.
+	Read(context.Context) (Message, int64, error)
+}
+
+// Writer abstracts the transport mechanics from the JSON RPC protocol.
+// A Conn writes messages using the writer it was provided on construction,
+// and assumes that each call to Write fully transfers a single message,
+// or returns an error.
+// A writer is not safe for concurrent use, it is expected it will be used by
+// a single Conn in a safe manner.
+type Writer interface {
+	// Write sends a message to the stream.
+	Write(context.Context, Message) (int64, error)
+}
+
+// Framer wraps low level byte readers and writers into jsonrpc2 message
+// readers and writers.
+// It is responsible for the framing and encoding of messages into wire form.
+type Framer interface {
+	// Reader wraps a byte reader into a message reader.
+	Reader(rw io.Reader) Reader
+	// Writer wraps a byte writer into a message writer.
+	Writer(rw io.Writer) Writer
+}
+
+// RawFramer returns a new Framer.
+// The messages are sent with no wrapping, and rely on json decode consistency
+// to determine message boundaries.
+func RawFramer() Framer { return rawFramer{} }
+
+type rawFramer struct{}
+type rawReader struct{ in *json.Decoder }
+type rawWriter struct{ out io.Writer }
+
+func (rawFramer) Reader(rw io.Reader) Reader {
+	return &rawReader{in: json.NewDecoder(rw)}
+}
+
+func (rawFramer) Writer(rw io.Writer) Writer {
+	return &rawWriter{out: rw}
+}
+
+func (r *rawReader) Read(ctx context.Context) (Message, int64, error) {
+	select {
+	case <-ctx.Done():
+		return nil, 0, ctx.Err()
+	default:
+	}
+	var raw json.RawMessage
+	if err := r.in.Decode(&raw); err != nil {
+		return nil, 0, err
+	}
+	msg, err := DecodeMessage(raw)
+	return msg, int64(len(raw)), err
+}
+
+func (w *rawWriter) Write(ctx context.Context, msg Message) (int64, error) {
+	select {
+	case <-ctx.Done():
+		return 0, ctx.Err()
+	default:
+	}
+	data, err := EncodeMessage(msg)
+	if err != nil {
+		return 0, errors.Errorf("marshaling message: %v", err)
+	}
+	n, err := w.out.Write(data)
+	return int64(n), err
+}
+
+// HeaderFramer returns a new Framer.
+// The messages are sent with HTTP content length and MIME type headers.
+// This is the format used by LSP and others.
+func HeaderFramer() Framer { return headerFramer{} }
+
+type headerFramer struct{}
+type headerReader struct{ in *bufio.Reader }
+type headerWriter struct{ out io.Writer }
+
+func (headerFramer) Reader(rw io.Reader) Reader {
+	return &headerReader{in: bufio.NewReader(rw)}
+}
+
+func (headerFramer) Writer(rw io.Writer) Writer {
+	return &headerWriter{out: rw}
+}
+
+func (r *headerReader) Read(ctx context.Context) (Message, int64, error) {
+	select {
+	case <-ctx.Done():
+		return nil, 0, ctx.Err()
+	default:
+	}
+	var total, length int64
+	// read the header, stop on the first empty line
+	for {
+		line, err := r.in.ReadString('\n')
+		total += int64(len(line))
+		if err != nil {
+			return nil, total, errors.Errorf("failed reading header line: %w", err)
+		}
+		line = strings.TrimSpace(line)
+		// check we have a header line
+		if line == "" {
+			break
+		}
+		colon := strings.IndexRune(line, ':')
+		if colon < 0 {
+			return nil, total, errors.Errorf("invalid header line %q", line)
+		}
+		name, value := line[:colon], strings.TrimSpace(line[colon+1:])
+		switch name {
+		case "Content-Length":
+			if length, err = strconv.ParseInt(value, 10, 32); err != nil {
+				return nil, total, errors.Errorf("failed parsing Content-Length: %v", value)
+			}
+			if length <= 0 {
+				return nil, total, errors.Errorf("invalid Content-Length: %v", length)
+			}
+		default:
+			// ignoring unknown headers
+		}
+	}
+	if length == 0 {
+		return nil, total, errors.Errorf("missing Content-Length header")
+	}
+	data := make([]byte, length)
+	n, err := io.ReadFull(r.in, data)
+	total += int64(n)
+	if err != nil {
+		return nil, total, err
+	}
+	msg, err := DecodeMessage(data)
+	return msg, total, err
+}
+
+func (w *headerWriter) Write(ctx context.Context, msg Message) (int64, error) {
+	select {
+	case <-ctx.Done():
+		return 0, ctx.Err()
+	default:
+	}
+	data, err := EncodeMessage(msg)
+	if err != nil {
+		return 0, errors.Errorf("marshaling message: %v", err)
+	}
+	n, err := fmt.Fprintf(w.out, "Content-Length: %v\r\n\r\n", len(data))
+	total := int64(n)
+	if err == nil {
+		n, err = w.out.Write(data)
+		total += int64(n)
+	}
+	return total, err
+}

diff --git a/jsonrpc2/jsonrpc2.go b/jsonrpc2/jsonrpc2.go
new file mode 100644
index 0000000..4e853d5
--- /dev/null
+++ b/jsonrpc2/jsonrpc2.go

@@ -0,0 +1,99 @@
+// Copyright 2018 The Go 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 jsonrpc2 is a minimal implementation of the JSON RPC 2 spec.
+// https://www.jsonrpc.org/specification
+// It is intended to be compatible with other implementations at the wire level.
+package jsonrpc2
+
+import (
+	"context"
+	"errors"
+)
+
+var (
+	// ErrIdleTimeout is returned when serving timed out waiting for new connections.
+	ErrIdleTimeout = errors.New("timed out waiting for new connections")
+	// ErrNotHandled is returned from a handler to indicate it did not handle the
+	// message.
+	ErrNotHandled = errors.New("JSON RPC not handled")
+	// ErrAsyncResponse is returned from a handler to indicate it will generate a
+	// response asynchronously.
+	ErrAsyncResponse = errors.New("JSON RPC asynchronous response")
+)
+
+// Preempter handles messages on a connection before they are queued to the main
+// handler.
+// Primarily this is used for cancel handlers or notifications for which out of
+// order processing is not an issue.
+type Preempter interface {
+	// Preempt is invoked for each incoming request before it is queued.
+	// If the request is a call, it must return a value or an error for the reply.
+	// Preempt should not block or start any new messages on the connection.
+	Preempt(ctx context.Context, req *Request) (interface{}, error)
+}
+
+// Handler handles messages on a connection.
+type Handler interface {
+	// Handle is invoked for each incoming request.
+	// If the request is a call, it must return a value or an error for the reply.
+	Handle(ctx context.Context, req *Request) (interface{}, error)
+}
+
+type defaultHandler struct{}
+
+func (defaultHandler) Preempt(context.Context, *Request) (interface{}, error) {
+	return nil, ErrNotHandled
+}
+
+func (defaultHandler) Handle(context.Context, *Request) (interface{}, error) {
+	return nil, ErrNotHandled
+}
+
+type HandlerFunc func(ctx context.Context, req *Request) (interface{}, error)
+
+func (f HandlerFunc) Handle(ctx context.Context, req *Request) (interface{}, error) {
+	return f(ctx, req)
+}
+
+// async is a small helper for things with an asynchronous result that you can
+// wait for.
+type async struct {
+	ready  chan struct{}
+	errBox chan error
+}
+
+func (a *async) init() {
+	a.ready = make(chan struct{})
+	a.errBox = make(chan error, 1)
+	a.errBox <- nil
+}
+
+func (a *async) done() {
+	close(a.ready)
+}
+
+func (a *async) isDone() bool {
+	select {
+	case <-a.ready:
+		return true
+	default:
+		return false
+	}
+}
+
+func (a *async) wait() error {
+	<-a.ready
+	err := <-a.errBox
+	a.errBox <- err
+	return err
+}
+
+func (a *async) setError(err error) {
+	storedErr := <-a.errBox
+	if storedErr == nil {
+		storedErr = err
+	}
+	a.errBox <- storedErr
+}

diff --git a/jsonrpc2/jsonrpc2_test.go b/jsonrpc2/jsonrpc2_test.go
new file mode 100644
index 0000000..e0ffe97
--- /dev/null
+++ b/jsonrpc2/jsonrpc2_test.go

@@ -0,0 +1,389 @@
+// Copyright 2018 The Go 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 jsonrpc2_test
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"path"
+	"reflect"
+	"testing"
+	"time"
+
+	"golang.org/x/exp/event/adapter/eventtest"
+	"golang.org/x/exp/internal/stack/stacktest"
+	"golang.org/x/exp/jsonrpc2"
+	errors "golang.org/x/xerrors"
+)
+
+var callTests = []invoker{
+	call{"no_args", nil, true},
+	call{"one_string", "fish", "got:fish"},
+	call{"one_number", 10, "got:10"},
+	call{"join", []string{"a", "b", "c"}, "a/b/c"},
+	sequence{"notify", []invoker{
+		notify{"set", 3},
+		notify{"add", 5},
+		call{"get", nil, 8},
+	}},
+	sequence{"preempt", []invoker{
+		async{"a", "wait", "a"},
+		notify{"unblock", "a"},
+		collect{"a", true, false},
+	}},
+	sequence{"basic cancel", []invoker{
+		async{"b", "wait", "b"},
+		cancel{"b"},
+		collect{"b", nil, true},
+	}},
+	sequence{"queue", []invoker{
+		async{"a", "wait", "a"},
+		notify{"set", 1},
+		notify{"add", 2},
+		notify{"add", 3},
+		notify{"add", 4},
+		call{"peek", nil, 0}, // accumulator will not have any adds yet
+		notify{"unblock", "a"},
+		collect{"a", true, false},
+		call{"get", nil, 10}, // accumulator now has all the adds
+	}},
+	sequence{"fork", []invoker{
+		async{"a", "fork", "a"},
+		notify{"set", 1},
+		notify{"add", 2},
+		notify{"add", 3},
+		notify{"add", 4},
+		call{"get", nil, 10}, // fork will not have blocked the adds
+		notify{"unblock", "a"},
+		collect{"a", true, false},
+	}},
+}
+
+type binder struct {
+	framer  jsonrpc2.Framer
+	runTest func(*handler)
+}
+
+type handler struct {
+	conn        *jsonrpc2.Connection
+	accumulator int
+	waitersBox  chan map[string]chan struct{}
+	calls       map[string]*jsonrpc2.AsyncCall
+}
+
+type invoker interface {
+	Name() string
+	Invoke(t *testing.T, ctx context.Context, h *handler)
+}
+
+type notify struct {
+	method string
+	params interface{}
+}
+
+type call struct {
+	method string
+	params interface{}
+	expect interface{}
+}
+
+type async struct {
+	name   string
+	method string
+	params interface{}
+}
+
+type collect struct {
+	name   string
+	expect interface{}
+	fails  bool
+}
+
+type cancel struct {
+	name string
+}
+
+type sequence struct {
+	name  string
+	tests []invoker
+}
+
+type echo call
+
+type cancelParams struct{ ID int64 }
+
+func TestConnectionRaw(t *testing.T) {
+	testConnection(t, jsonrpc2.RawFramer())
+}
+
+func TestConnectionHeader(t *testing.T) {
+	testConnection(t, jsonrpc2.HeaderFramer())
+}
+
+func testConnection(t *testing.T, framer jsonrpc2.Framer) {
+	stacktest.NoLeak(t)
+	ctx := eventtest.NewContext(context.Background(), t)
+	listener, err := jsonrpc2.NetPipe(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	server, err := jsonrpc2.Serve(ctx, listener, binder{framer, nil})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		listener.Close()
+		server.Wait()
+	}()
+
+	for _, test := range callTests {
+		t.Run(test.Name(), func(t *testing.T) {
+			client, err := jsonrpc2.Dial(ctx,
+				listener.Dialer(), binder{framer, func(h *handler) {
+					defer h.conn.Close()
+					ctx := eventtest.NewContext(ctx, t)
+					test.Invoke(t, ctx, h)
+					if call, ok := test.(*call); ok {
+						// also run all simple call tests in echo mode
+						(*echo)(call).Invoke(t, ctx, h)
+					}
+				}})
+			if err != nil {
+				t.Fatal(err)
+			}
+			client.Wait()
+		})
+	}
+}
+
+func (test notify) Name() string { return test.method }
+func (test notify) Invoke(t *testing.T, ctx context.Context, h *handler) {
+	if err := h.conn.Notify(ctx, test.method, test.params); err != nil {
+		t.Fatalf("%v:Notify failed: %v", test.method, err)
+	}
+}
+
+func (test call) Name() string { return test.method }
+func (test call) Invoke(t *testing.T, ctx context.Context, h *handler) {
+	results := newResults(test.expect)
+	if err := h.conn.Call(ctx, test.method, test.params).Await(ctx, results); err != nil {
+		t.Fatalf("%v:Call failed: %v", test.method, err)
+	}
+	verifyResults(t, test.method, results, test.expect)
+}
+
+func (test echo) Invoke(t *testing.T, ctx context.Context, h *handler) {
+	results := newResults(test.expect)
+	if err := h.conn.Call(ctx, "echo", []interface{}{test.method, test.params}).Await(ctx, results); err != nil {
+		t.Fatalf("%v:Echo failed: %v", test.method, err)
+	}
+	verifyResults(t, test.method, results, test.expect)
+}
+
+func (test async) Name() string { return test.name }
+func (test async) Invoke(t *testing.T, ctx context.Context, h *handler) {
+	h.calls[test.name] = h.conn.Call(ctx, test.method, test.params)
+}
+
+func (test collect) Name() string { return test.name }
+func (test collect) Invoke(t *testing.T, ctx context.Context, h *handler) {
+	o := h.calls[test.name]
+	results := newResults(test.expect)
+	err := o.Await(ctx, results)
+	switch {
+	case test.fails && err == nil:
+		t.Fatalf("%v:Collect was supposed to fail", test.name)
+	case !test.fails && err != nil:
+		t.Fatalf("%v:Collect failed: %v", test.name, err)
+	}
+	verifyResults(t, test.name, results, test.expect)
+}
+
+func (test cancel) Name() string { return test.name }
+func (test cancel) Invoke(t *testing.T, ctx context.Context, h *handler) {
+	o := h.calls[test.name]
+	if err := h.conn.Notify(ctx, "cancel", &cancelParams{o.ID().Raw().(int64)}); err != nil {
+		t.Fatalf("%v:Collect failed: %v", test.name, err)
+	}
+}
+
+func (test sequence) Name() string { return test.name }
+func (test sequence) Invoke(t *testing.T, ctx context.Context, h *handler) {
+	for _, child := range test.tests {
+		child.Invoke(t, ctx, h)
+	}
+}
+
+// newResults makes a new empty copy of the expected type to put the results into
+func newResults(expect interface{}) interface{} {
+	switch e := expect.(type) {
+	case []interface{}:
+		var r []interface{}
+		for _, v := range e {
+			r = append(r, reflect.New(reflect.TypeOf(v)).Interface())
+		}
+		return r
+	case nil:
+		return nil
+	default:
+		return reflect.New(reflect.TypeOf(expect)).Interface()
+	}
+}
+
+// verifyResults compares the results to the expected values
+func verifyResults(t *testing.T, method string, results interface{}, expect interface{}) {
+	if expect == nil {
+		if results != nil {
+			t.Errorf("%v:Got results %+v where none expeted", method, expect)
+		}
+		return
+	}
+	val := reflect.Indirect(reflect.ValueOf(results)).Interface()
+	if !reflect.DeepEqual(val, expect) {
+		t.Errorf("%v:Results are incorrect, got %+v expect %+v", method, val, expect)
+	}
+}
+
+func (b binder) Bind(ctx context.Context, conn *jsonrpc2.Connection) (jsonrpc2.ConnectionOptions, error) {
+	h := &handler{
+		conn:       conn,
+		waitersBox: make(chan map[string]chan struct{}, 1),
+		calls:      make(map[string]*jsonrpc2.AsyncCall),
+	}
+	h.waitersBox <- make(map[string]chan struct{})
+	if b.runTest != nil {
+		go b.runTest(h)
+	}
+	return jsonrpc2.ConnectionOptions{
+		Framer:    b.framer,
+		Preempter: h,
+		Handler:   h,
+	}, nil
+}
+
+func (h *handler) waiter(name string) chan struct{} {
+	waiters := <-h.waitersBox
+	defer func() { h.waitersBox <- waiters }()
+	waiter, found := waiters[name]
+	if !found {
+		waiter = make(chan struct{})
+		waiters[name] = waiter
+	}
+	return waiter
+}
+
+func (h *handler) Preempt(ctx context.Context, req *jsonrpc2.Request) (interface{}, error) {
+	switch req.Method {
+	case "unblock":
+		var name string
+		if err := json.Unmarshal(req.Params, &name); err != nil {
+			return nil, errors.Errorf("%w: %s", jsonrpc2.ErrParse, err)
+		}
+		close(h.waiter(name))
+		return nil, nil
+	case "peek":
+		if len(req.Params) > 0 {
+			return nil, errors.Errorf("%w: expected no params", jsonrpc2.ErrInvalidParams)
+		}
+		return h.accumulator, nil
+	case "cancel":
+		var params cancelParams
+		if err := json.Unmarshal(req.Params, &params); err != nil {
+			return nil, errors.Errorf("%w: %s", jsonrpc2.ErrParse, err)
+		}
+		h.conn.Cancel(jsonrpc2.Int64ID(params.ID))
+		return nil, nil
+	default:
+		return nil, jsonrpc2.ErrNotHandled
+	}
+}
+
+func (h *handler) Handle(ctx context.Context, req *jsonrpc2.Request) (interface{}, error) {
+	switch req.Method {
+	case "no_args":
+		if len(req.Params) > 0 {
+			return nil, errors.Errorf("%w: expected no params", jsonrpc2.ErrInvalidParams)
+		}
+		return true, nil
+	case "one_string":
+		var v string
+		if err := json.Unmarshal(req.Params, &v); err != nil {
+			return nil, errors.Errorf("%w: %s", jsonrpc2.ErrParse, err)
+		}
+		return "got:" + v, nil
+	case "one_number":
+		var v int
+		if err := json.Unmarshal(req.Params, &v); err != nil {
+			return nil, errors.Errorf("%w: %s", jsonrpc2.ErrParse, err)
+		}
+		return fmt.Sprintf("got:%d", v), nil
+	case "set":
+		var v int
+		if err := json.Unmarshal(req.Params, &v); err != nil {
+			return nil, errors.Errorf("%w: %s", jsonrpc2.ErrParse, err)
+		}
+		h.accumulator = v
+		return nil, nil
+	case "add":
+		var v int
+		if err := json.Unmarshal(req.Params, &v); err != nil {
+			return nil, errors.Errorf("%w: %s", jsonrpc2.ErrParse, err)
+		}
+		h.accumulator += v
+		return nil, nil
+	case "get":
+		if len(req.Params) > 0 {
+			return nil, errors.Errorf("%w: expected no params", jsonrpc2.ErrInvalidParams)
+		}
+		return h.accumulator, nil
+	case "join":
+		var v []string
+		if err := json.Unmarshal(req.Params, &v); err != nil {
+			return nil, errors.Errorf("%w: %s", jsonrpc2.ErrParse, err)
+		}
+		return path.Join(v...), nil
+	case "echo":
+		var v []interface{}
+		if err := json.Unmarshal(req.Params, &v); err != nil {
+			return nil, errors.Errorf("%w: %s", jsonrpc2.ErrParse, err)
+		}
+		var result interface{}
+		err := h.conn.Call(ctx, v[0].(string), v[1]).Await(ctx, &result)
+		return result, err
+	case "wait":
+		var name string
+		if err := json.Unmarshal(req.Params, &name); err != nil {
+			return nil, errors.Errorf("%w: %s", jsonrpc2.ErrParse, err)
+		}
+		select {
+		case <-h.waiter(name):
+			return true, nil
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-time.After(time.Second):
+			return nil, errors.Errorf("wait for %q timed out", name)
+		}
+	case "fork":
+		var name string
+		if err := json.Unmarshal(req.Params, &name); err != nil {
+			return nil, errors.Errorf("%w: %s", jsonrpc2.ErrParse, err)
+		}
+		waitFor := h.waiter(name)
+		go func() {
+			select {
+			case <-waitFor:
+				h.conn.Respond(req.ID, true, nil)
+			case <-ctx.Done():
+				h.conn.Respond(req.ID, nil, ctx.Err())
+			case <-time.After(time.Second):
+				h.conn.Respond(req.ID, nil, errors.Errorf("wait for %q timed out", name))
+			}
+		}()
+		return nil, jsonrpc2.ErrAsyncResponse
+	default:
+		return nil, jsonrpc2.ErrNotHandled
+	}
+}

diff --git a/jsonrpc2/messages.go b/jsonrpc2/messages.go
new file mode 100644
index 0000000..652ac81
--- /dev/null
+++ b/jsonrpc2/messages.go

@@ -0,0 +1,181 @@
+// Copyright 2018 The Go 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 jsonrpc2
+
+import (
+	"encoding/json"
+
+	errors "golang.org/x/xerrors"
+)
+
+// ID is a Request identifier.
+type ID struct {
+	value interface{}
+}
+
+// Message is the interface to all jsonrpc2 message types.
+// They share no common functionality, but are a closed set of concrete types
+// that are allowed to implement this interface. The message types are *Request
+// and *Response.
+type Message interface {
+	// marshal builds the wire form from the API form.
+	// It is private, which makes the set of Message implementations closed.
+	marshal(to *wireCombined)
+}
+
+// Request is a Message sent to a peer to request behavior.
+// If it has an ID it is a call, otherwise it is a notification.
+type Request struct {
+	// ID of this request, used to tie the Response back to the request.
+	// This will be nil for notifications.
+	ID ID
+	// Method is a string containing the method name to invoke.
+	Method string
+	// Params is either a struct or an array with the parameters of the method.
+	Params json.RawMessage
+}
+
+// Response is a Message used as a reply to a call Request.
+// It will have the same ID as the call it is a response to.
+type Response struct {
+	// result is the content of the response.
+	Result json.RawMessage
+	// err is set only if the call failed.
+	Error error
+	// id of the request this is a response to.
+	ID ID
+}
+
+// StringID creates a new string request identifier.
+func StringID(s string) ID { return ID{value: s} }
+
+// Int64ID creates a new integer request identifier.
+func Int64ID(i int64) ID { return ID{value: i} }
+
+// IsValid returns true if the ID is a valid identifier.
+// The default value for ID will return false.
+func (id ID) IsValid() bool { return id.value != nil }
+
+// Raw returns the underlying value of the ID.
+func (id ID) Raw() interface{} { return id.value }
+
+// NewNotification constructs a new Notification message for the supplied
+// method and parameters.
+func NewNotification(method string, params interface{}) (*Request, error) {
+	p, merr := marshalToRaw(params)
+	return &Request{Method: method, Params: p}, merr
+}
+
+// NewCall constructs a new Call message for the supplied ID, method and
+// parameters.
+func NewCall(id ID, method string, params interface{}) (*Request, error) {
+	p, merr := marshalToRaw(params)
+	return &Request{ID: id, Method: method, Params: p}, merr
+}
+
+func (msg *Request) IsCall() bool { return msg.ID.IsValid() }
+
+func (msg *Request) marshal(to *wireCombined) {
+	to.ID = msg.ID.value
+	to.Method = msg.Method
+	to.Params = msg.Params
+}
+
+// NewResponse constructs a new Response message that is a reply to the
+// supplied. If err is set result may be ignored.
+func NewResponse(id ID, result interface{}, rerr error) (*Response, error) {
+	r, merr := marshalToRaw(result)
+	return &Response{ID: id, Result: r, Error: rerr}, merr
+}
+
+func (msg *Response) marshal(to *wireCombined) {
+	to.ID = msg.ID.value
+	to.Error = toWireError(msg.Error)
+	to.Result = msg.Result
+}
+
+func toWireError(err error) *wireError {
+	if err == nil {
+		// no error, the response is complete
+		return nil
+	}
+	if err, ok := err.(*wireError); ok {
+		// already a wire error, just use it
+		return err
+	}
+	result := &wireError{Message: err.Error()}
+	var wrapped *wireError
+	if errors.As(err, &wrapped) {
+		// if we wrapped a wire error, keep the code from the wrapped error
+		// but the message from the outer error
+		result.Code = wrapped.Code
+	}
+	return result
+}
+
+func EncodeMessage(msg Message) ([]byte, error) {
+	wire := wireCombined{VersionTag: wireVersion}
+	msg.marshal(&wire)
+	data, err := json.Marshal(&wire)
+	if err != nil {
+		return data, errors.Errorf("marshaling jsonrpc message: %w", err)
+	}
+	return data, nil
+}
+
+func DecodeMessage(data []byte) (Message, error) {
+	msg := wireCombined{}
+	if err := json.Unmarshal(data, &msg); err != nil {
+		return nil, errors.Errorf("unmarshaling jsonrpc message: %w", err)
+	}
+	if msg.VersionTag != wireVersion {
+		return nil, errors.Errorf("invalid message version tag %s expected %s", msg.VersionTag, wireVersion)
+	}
+	id := ID{}
+	switch v := msg.ID.(type) {
+	case nil:
+	case float64:
+		// coerce the id type to int64 if it is float64, the spec does not allow fractional parts
+		id = Int64ID(int64(v))
+	case int64:
+		id = Int64ID(v)
+	case string:
+		id = StringID(v)
+	default:
+		return nil, errors.Errorf("invalid message id type <%T>%v", v, v)
+	}
+	if msg.Method != "" {
+		// has a method, must be a call
+		return &Request{
+			Method: msg.Method,
+			ID:     id,
+			Params: msg.Params,
+		}, nil
+	}
+	// no method, should be a response
+	if !id.IsValid() {
+		return nil, ErrInvalidRequest
+	}
+	resp := &Response{
+		ID:     id,
+		Result: msg.Result,
+	}
+	// we have to check if msg.Error is nil to avoid a typed error
+	if msg.Error != nil {
+		resp.Error = msg.Error
+	}
+	return resp, nil
+}
+
+func marshalToRaw(obj interface{}) (json.RawMessage, error) {
+	if obj == nil {
+		return nil, nil
+	}
+	data, err := json.Marshal(obj)
+	if err != nil {
+		return nil, err
+	}
+	return json.RawMessage(data), nil
+}

diff --git a/jsonrpc2/net.go b/jsonrpc2/net.go
new file mode 100644
index 0000000..c8cfaab
--- /dev/null
+++ b/jsonrpc2/net.go

@@ -0,0 +1,129 @@
+// Copyright 2018 The Go 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 jsonrpc2
+
+import (
+	"context"
+	"io"
+	"net"
+	"os"
+	"time"
+)
+
+// This file contains implementations of the transport primitives that use the standard network
+// package.
+
+// NetListenOptions is the optional arguments to the NetListen function.
+type NetListenOptions struct {
+	NetListenConfig net.ListenConfig
+	NetDialer       net.Dialer
+}
+
+// NetListener returns a new Listener that listents on a socket using the net package.
+func NetListener(ctx context.Context, network, address string, options NetListenOptions) (Listener, error) {
+	ln, err := options.NetListenConfig.Listen(ctx, network, address)
+	if err != nil {
+		return nil, err
+	}
+	return &netListener{net: ln}, nil
+}
+
+// netListener is the implementation of Listener for connections made using the net package.
+type netListener struct {
+	net net.Listener
+}
+
+// Accept blocks waiting for an incoming connection to the listener.
+func (l *netListener) Accept(ctx context.Context) (io.ReadWriteCloser, error) {
+	return l.net.Accept()
+}
+
+// Close will cause the listener to stop listening. It will not close any connections that have
+// already been accepted.
+func (l *netListener) Close() error {
+	addr := l.net.Addr()
+	err := l.net.Close()
+	if addr.Network() == "unix" {
+		rerr := os.Remove(addr.String())
+		if rerr != nil && err == nil {
+			err = rerr
+		}
+	}
+	return err
+}
+
+// Dialer returns a dialer that can be used to connect to the listener.
+func (l *netListener) Dialer() Dialer {
+	return NetDialer(l.net.Addr().Network(), l.net.Addr().String(), net.Dialer{
+		Timeout: 5 * time.Second,
+	})
+}
+
+// NetDialer returns a Dialer using the supplied standard network dialer.
+func NetDialer(network, address string, nd net.Dialer) Dialer {
+	return &netDialer{
+		network: network,
+		address: address,
+		dialer:  nd,
+	}
+}
+
+type netDialer struct {
+	network string
+	address string
+	dialer  net.Dialer
+}
+
+func (n *netDialer) Dial(ctx context.Context) (io.ReadWriteCloser, error) {
+	return n.dialer.DialContext(ctx, n.network, n.address)
+}
+
+// NetPipe returns a new Listener that listens using net.Pipe.
+// It is only possibly to connect to it using the Dialier returned by the
+// Dialer method, each call to that method will generate a new pipe the other
+// side of which will be returnd from the Accept call.
+func NetPipe(ctx context.Context) (Listener, error) {
+	return &netPiper{
+		done:   make(chan struct{}),
+		dialed: make(chan io.ReadWriteCloser),
+	}, nil
+}
+
+// netPiper is the implementation of Listener build on top of net.Pipes.
+type netPiper struct {
+	done   chan struct{}
+	dialed chan io.ReadWriteCloser
+}
+
+// Accept blocks waiting for an incoming connection to the listener.
+func (l *netPiper) Accept(ctx context.Context) (io.ReadWriteCloser, error) {
+	// block until we have a listener, or are closed or cancelled
+	select {
+	case rwc := <-l.dialed:
+		return rwc, nil
+	case <-l.done:
+		return nil, io.EOF
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+// Close will cause the listener to stop listening. It will not close any connections that have
+// already been accepted.
+func (l *netPiper) Close() error {
+	// unblock any accept calls that are pending
+	close(l.done)
+	return nil
+}
+
+func (l *netPiper) Dialer() Dialer {
+	return l
+}
+
+func (l *netPiper) Dial(ctx context.Context) (io.ReadWriteCloser, error) {
+	client, server := net.Pipe()
+	l.dialed <- server
+	return client, nil
+}

diff --git a/jsonrpc2/serve.go b/jsonrpc2/serve.go
new file mode 100644
index 0000000..f3b78f5
--- /dev/null
+++ b/jsonrpc2/serve.go

@@ -0,0 +1,283 @@
+// Copyright 2020 The Go 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 jsonrpc2
+
+import (
+	"context"
+	"io"
+	"runtime"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+
+	errors "golang.org/x/xerrors"
+)
+
+// Listener is implemented by protocols to accept new inbound connections.
+type Listener interface {
+	// Accept an inbound connection to a server.
+	// It must block until an inbound connection is made, or the listener is
+	// shut down.
+	Accept(context.Context) (io.ReadWriteCloser, error)
+
+	// Close is used to ask a listener to stop accepting new connections.
+	Close() error
+
+	// Dialer returns a dialer that can be used to connect to this listener
+	// locally.
+	// If a listener does not implement this it will return a nil.
+	Dialer() Dialer
+}
+
+// Dialer is used by clients to dial a server.
+type Dialer interface {
+	// Dial returns a new communication byte stream to a listening server.
+	Dial(ctx context.Context) (io.ReadWriteCloser, error)
+}
+
+// Server is a running server that is accepting incoming connections.
+type Server struct {
+	listener Listener
+	binder   Binder
+	async    async
+}
+
+// Dial uses the dialer to make a new connection, wraps the returned
+// reader and writer using the framer to make a stream, and then builds
+// a connection on top of that stream using the binder.
+func Dial(ctx context.Context, dialer Dialer, binder Binder) (*Connection, error) {
+	// dial a server
+	rwc, err := dialer.Dial(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return newConnection(ctx, rwc, binder)
+}
+
+// Serve starts a new server listening for incoming connections and returns
+// it.
+// This returns a fully running and connected server, it does not block on
+// the listener.
+// You can call Wait to block on the server, or Shutdown to get the sever to
+// terminate gracefully.
+// To notice incoming connections, use an intercepting Binder.
+func Serve(ctx context.Context, listener Listener, binder Binder) (*Server, error) {
+	server := &Server{
+		listener: listener,
+		binder:   binder,
+	}
+	server.async.init()
+	go server.run(ctx)
+	return server, nil
+}
+
+// Wait returns only when the server has shut down.
+func (s *Server) Wait() error {
+	return s.async.wait()
+}
+
+// run accepts incoming connections from the listener,
+// If IdleTimeout is non-zero, run exits after there are no clients for this
+// duration, otherwise it exits only on error.
+func (s *Server) run(ctx context.Context) {
+	defer s.async.done()
+	var activeConns []*Connection
+	for {
+		// we never close the accepted connection, we rely on the other end
+		// closing or the socket closing itself naturally
+		rwc, err := s.listener.Accept(ctx)
+		if err != nil {
+			if !isClosingError(err) {
+				s.async.setError(err)
+			}
+			// we are done generating new connections for good
+			break
+		}
+
+		// see if any connections were closed while we were waiting
+		activeConns = onlyActive(activeConns)
+
+		// a new inbound connection,
+		conn, err := newConnection(ctx, rwc, s.binder)
+		if err != nil {
+			if !isClosingError(err) {
+				s.async.setError(err)
+			}
+			continue
+		}
+		activeConns = append(activeConns, conn)
+	}
+
+	// wait for all active conns to finish
+	for _, c := range activeConns {
+		c.Wait()
+	}
+}
+
+func onlyActive(conns []*Connection) []*Connection {
+	i := 0
+	for _, c := range conns {
+		if !c.async.isDone() {
+			conns[i] = c
+			i++
+		}
+	}
+	// trim the slice down
+	return conns[:i]
+}
+
+// isClosingError reports if the error occurs normally during the process of
+// closing a network connection. It uses imperfect heuristics that err on the
+// side of false negatives, and should not be used for anything critical.
+func isClosingError(err error) bool {
+	if err == nil {
+		return false
+	}
+	// Fully unwrap the error, so the following tests work.
+	for wrapped := err; wrapped != nil; wrapped = errors.Unwrap(err) {
+		err = wrapped
+	}
+
+	// Was it based on an EOF error?
+	if err == io.EOF {
+		return true
+	}
+
+	// Was it based on a closed pipe?
+	if err == io.ErrClosedPipe {
+		return true
+	}
+
+	// Per https://github.com/golang/go/issues/4373, this error string should not
+	// change. This is not ideal, but since the worst that could happen here is
+	// some superfluous logging, it is acceptable.
+	if err.Error() == "use of closed network connection" {
+		return true
+	}
+
+	if runtime.GOOS == "plan9" {
+		// Error reading from a closed connection.
+		if err == syscall.EINVAL {
+			return true
+		}
+		// Error trying to accept a new connection from a closed listener.
+		if strings.HasSuffix(err.Error(), " listen hungup") {
+			return true
+		}
+	}
+	return false
+}
+
+// NewIdleListener wraps a listener with an idle timeout.
+// When there are no active connections for at least the timeout duration a
+// call to accept will fail with ErrIdleTimeout.
+func NewIdleListener(timeout time.Duration, wrap Listener) Listener {
+	l := &idleListener{
+		timeout:    timeout,
+		wrapped:    wrap,
+		newConns:   make(chan *idleCloser),
+		closed:     make(chan struct{}),
+		wasTimeout: make(chan struct{}),
+	}
+	go l.run()
+	return l
+}
+
+type idleListener struct {
+	wrapped    Listener
+	timeout    time.Duration
+	newConns   chan *idleCloser
+	closed     chan struct{}
+	wasTimeout chan struct{}
+	closeOnce  sync.Once
+}
+
+type idleCloser struct {
+	wrapped   io.ReadWriteCloser
+	closed    chan struct{}
+	closeOnce sync.Once
+}
+
+func (c *idleCloser) Read(p []byte) (int, error) {
+	n, err := c.wrapped.Read(p)
+	if err != nil && isClosingError(err) {
+		c.closeOnce.Do(func() { close(c.closed) })
+	}
+	return n, err
+}
+
+func (c *idleCloser) Write(p []byte) (int, error) {
+	// we do not close on write failure, we rely on the wrapped writer to do that
+	// if it is appropriate, which we will detect in the next read.
+	return c.wrapped.Write(p)
+}
+
+func (c *idleCloser) Close() error {
+	// we rely on closing the wrapped stream to signal to the next read that we
+	// are closed, rather than triggering the closed signal directly
+	return c.wrapped.Close()
+}
+
+func (l *idleListener) Accept(ctx context.Context) (io.ReadWriteCloser, error) {
+	rwc, err := l.wrapped.Accept(ctx)
+	if err != nil {
+		if isClosingError(err) {
+			// underlying listener was closed
+			l.closeOnce.Do(func() { close(l.closed) })
+			// was it closed because of the idle timeout?
+			select {
+			case <-l.wasTimeout:
+				err = ErrIdleTimeout
+			default:
+			}
+		}
+		return nil, err
+	}
+	conn := &idleCloser{
+		wrapped: rwc,
+		closed:  make(chan struct{}),
+	}
+	l.newConns <- conn
+	return conn, err
+}
+
+func (l *idleListener) Close() error {
+	defer l.closeOnce.Do(func() { close(l.closed) })
+	return l.wrapped.Close()
+}
+
+func (l *idleListener) Dialer() Dialer {
+	return l.wrapped.Dialer()
+}
+
+func (l *idleListener) run() {
+	var conns []*idleCloser
+	for {
+		var firstClosed chan struct{} // left at nil if there are no active conns
+		var timeout <-chan time.Time  // left at nil if there are  active conns
+		if len(conns) > 0 {
+			firstClosed = conns[0].closed
+		} else {
+			timeout = time.After(l.timeout)
+		}
+		select {
+		case <-l.closed:
+			// the main listener closed, no need to keep going
+			return
+		case conn := <-l.newConns:
+			// a new conn arrived, add it to the list
+			conns = append(conns, conn)
+		case <-timeout:
+			// we timed out, only happens when there are no active conns
+			// close the underlying listener, and allow the normal closing process to happen
+			close(l.wasTimeout)
+			l.wrapped.Close()
+		case <-firstClosed:
+			// a conn closed, remove it from the active list
+			conns = conns[:copy(conns, conns[1:])]
+		}
+	}
+}

diff --git a/jsonrpc2/serve_test.go b/jsonrpc2/serve_test.go
new file mode 100644
index 0000000..475a966
--- /dev/null
+++ b/jsonrpc2/serve_test.go

@@ -0,0 +1,144 @@
+// Copyright 2020 The Go 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 jsonrpc2_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"golang.org/x/exp/internal/stack/stacktest"
+	"golang.org/x/exp/jsonrpc2"
+)
+
+func TestIdleTimeout(t *testing.T) {
+	stacktest.NoLeak(t)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	listener, err := jsonrpc2.NetListener(ctx, "tcp", "localhost:0", jsonrpc2.NetListenOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	listener = jsonrpc2.NewIdleListener(100*time.Millisecond, listener)
+	defer listener.Close()
+	server, err := jsonrpc2.Serve(ctx, listener, jsonrpc2.ConnectionOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	connect := func() *jsonrpc2.Connection {
+		client, err := jsonrpc2.Dial(ctx,
+			listener.Dialer(),
+			jsonrpc2.ConnectionOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return client
+	}
+	// Exercise some connection/disconnection patterns, and then assert that when
+	// our timer fires, the server exits.
+	conn1 := connect()
+	conn2 := connect()
+	if err := conn1.Close(); err != nil {
+		t.Fatalf("conn1.Close failed with error: %v", err)
+	}
+	if err := conn2.Close(); err != nil {
+		t.Fatalf("conn2.Close failed with error: %v", err)
+	}
+	conn3 := connect()
+	if err := conn3.Close(); err != nil {
+		t.Fatalf("conn3.Close failed with error: %v", err)
+	}
+
+	serverError := server.Wait()
+
+	if !errors.Is(serverError, jsonrpc2.ErrIdleTimeout) {
+		t.Errorf("run() returned error %v, want %v", serverError, jsonrpc2.ErrIdleTimeout)
+	}
+}
+
+type msg struct {
+	Msg string
+}
+
+type fakeHandler struct{}
+
+func (fakeHandler) Handle(ctx context.Context, req *jsonrpc2.Request) (interface{}, error) {
+	switch req.Method {
+	case "ping":
+		return &msg{"pong"}, nil
+	default:
+		return nil, jsonrpc2.ErrNotHandled
+	}
+}
+
+func TestServe(t *testing.T) {
+	stacktest.NoLeak(t)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	tests := []struct {
+		name    string
+		factory func(context.Context) (jsonrpc2.Listener, error)
+	}{
+		{"tcp", func(ctx context.Context) (jsonrpc2.Listener, error) {
+			return jsonrpc2.NetListener(ctx, "tcp", "localhost:0", jsonrpc2.NetListenOptions{})
+		}},
+		{"pipe", func(ctx context.Context) (jsonrpc2.Listener, error) {
+			return jsonrpc2.NetPipe(ctx)
+		}},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fake, err := test.factory(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+			conn, shutdown, err := newFake(t, ctx, fake)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer shutdown()
+			var got msg
+			if err := conn.Call(ctx, "ping", &msg{"ting"}).Await(ctx, &got); err != nil {
+				t.Fatal(err)
+			}
+			if want := "pong"; got.Msg != want {
+				t.Errorf("conn.Call(...): returned %q, want %q", got, want)
+			}
+		})
+	}
+}
+
+func newFake(t *testing.T, ctx context.Context, l jsonrpc2.Listener) (*jsonrpc2.Connection, func(), error) {
+	l = jsonrpc2.NewIdleListener(100*time.Millisecond, l)
+	server, err := jsonrpc2.Serve(ctx, l, jsonrpc2.ConnectionOptions{
+		Handler: fakeHandler{},
+	})
+	if err != nil {
+		return nil, nil, err
+	}
+
+	client, err := jsonrpc2.Dial(ctx,
+		l.Dialer(),
+		jsonrpc2.ConnectionOptions{
+			Handler: fakeHandler{},
+		})
+	if err != nil {
+		return nil, nil, err
+	}
+	return client, func() {
+		if err := l.Close(); err != nil {
+			t.Fatal(err)
+		}
+		if err := client.Close(); err != nil {
+			t.Fatal(err)
+		}
+		server.Wait()
+	}, nil
+}

diff --git a/jsonrpc2/wire.go b/jsonrpc2/wire.go
new file mode 100644
index 0000000..97b1ae8
--- /dev/null
+++ b/jsonrpc2/wire.go

@@ -0,0 +1,74 @@
+// Copyright 2018 The Go 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 jsonrpc2
+
+import (
+	"encoding/json"
+)
+
+// This file contains the go forms of the wire specification.
+// see http://www.jsonrpc.org/specification for details
+
+var (
+	// ErrUnknown should be used for all non coded errors.
+	ErrUnknown = NewError(-32001, "JSON RPC unknown error")
+	// ErrParse is used when invalid JSON was received by the server.
+	ErrParse = NewError(-32700, "JSON RPC parse error")
+	// ErrInvalidRequest is used when the JSON sent is not a valid Request object.
+	ErrInvalidRequest = NewError(-32600, "JSON RPC invalid request")
+	// ErrMethodNotFound should be returned by the handler when the method does
+	// not exist / is not available.
+	ErrMethodNotFound = NewError(-32601, "JSON RPC method not found")
+	// ErrInvalidParams should be returned by the handler when method
+	// parameter(s) were invalid.
+	ErrInvalidParams = NewError(-32602, "JSON RPC invalid params")
+	// ErrInternal indicates a failure to process a call correctly
+	ErrInternal = NewError(-32603, "JSON RPC internal error")
+
+	// The following errors are not part of the json specification, but
+	// compliant extensions specific to this implimentation.
+
+	// ErrServerOverloaded is returned when a message was refused due to a
+	// server being temporarily unable to accept any new messages.
+	ErrServerOverloaded = NewError(-32000, "JSON RPC overloaded")
+)
+
+const wireVersion = "2.0"
+
+// wireCombined has all the fields of both Request and Response.
+// We can decode this and then work out which it is.
+type wireCombined struct {
+	VersionTag string          `json:"jsonrpc"`
+	ID         interface{}     `json:"id,omitempty"`
+	Method     string          `json:"method,omitempty"`
+	Params     json.RawMessage `json:"params,omitempty"`
+	Result     json.RawMessage `json:"result,omitempty"`
+	Error      *wireError      `json:"error,omitempty"`
+}
+
+// wireError represents a structured error in a Response.
+type wireError struct {
+	// Code is an error code indicating the type of failure.
+	Code int64 `json:"code"`
+	// Message is a short description of the error.
+	Message string `json:"message"`
+	// Data is optional structured data containing additional information about the error.
+	Data json.RawMessage `json:"data,omitempty"`
+}
+
+// NewError returns an error that will encode on the wire correctly.
+// The standard codes are made available from this package, this function should
+// only be used to build errors for application specific codes as allowed by the
+// specification.
+func NewError(code int64, message string) error {
+	return &wireError{
+		Code:    code,
+		Message: message,
+	}
+}
+
+func (err *wireError) Error() string {
+	return err.Message
+}

diff --git a/jsonrpc2/wire_test.go b/jsonrpc2/wire_test.go
new file mode 100644
index 0000000..905687d
--- /dev/null
+++ b/jsonrpc2/wire_test.go

@@ -0,0 +1,118 @@
+// Copyright 2020 The Go 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 jsonrpc2_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"reflect"
+	"testing"
+
+	"golang.org/x/exp/jsonrpc2"
+)
+
+func TestWireMessage(t *testing.T) {
+	for _, test := range []struct {
+		name    string
+		msg     jsonrpc2.Message
+		encoded []byte
+	}{{
+		name:    "notification",
+		msg:     newNotification("alive", nil),
+		encoded: []byte(`{"jsonrpc":"2.0","method":"alive"}`),
+	}, {
+		name:    "call",
+		msg:     newCall("msg1", "ping", nil),
+		encoded: []byte(`{"jsonrpc":"2.0","id":"msg1","method":"ping"}`),
+	}, {
+		name:    "response",
+		msg:     newResponse("msg2", "pong", nil),
+		encoded: []byte(`{"jsonrpc":"2.0","id":"msg2","result":"pong"}`),
+	}, {
+		name:    "numerical id",
+		msg:     newCall(1, "poke", nil),
+		encoded: []byte(`{"jsonrpc":"2.0","id":1,"method":"poke"}`),
+	}, {
+		// originally reported in #39719, this checks that result is not present if
+		// it is an error response
+		name: "computing fix edits",
+		msg:  newResponse(3, nil, jsonrpc2.NewError(0, "computing fix edits")),
+		encoded: []byte(`{
+		"jsonrpc":"2.0",
+		"id":3,
+		"error":{
+			"code":0,
+			"message":"computing fix edits"
+		}
+	}`),
+	}} {
+		b, err := jsonrpc2.EncodeMessage(test.msg)
+		if err != nil {
+			t.Fatal(err)
+		}
+		checkJSON(t, b, test.encoded)
+		msg, err := jsonrpc2.DecodeMessage(test.encoded)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !reflect.DeepEqual(msg, test.msg) {
+			t.Errorf("decoded message does not match\nGot:\n%+#v\nWant:\n%+#v", msg, test.msg)
+		}
+	}
+}
+
+func newNotification(method string, params interface{}) jsonrpc2.Message {
+	msg, err := jsonrpc2.NewNotification(method, params)
+	if err != nil {
+		panic(err)
+	}
+	return msg
+}
+
+func newID(id interface{}) jsonrpc2.ID {
+	switch v := id.(type) {
+	case nil:
+		return jsonrpc2.ID{}
+	case string:
+		return jsonrpc2.StringID(v)
+	case int:
+		return jsonrpc2.Int64ID(int64(v))
+	case int64:
+		return jsonrpc2.Int64ID(v)
+	default:
+		panic("invalid ID type")
+	}
+}
+
+func newCall(id interface{}, method string, params interface{}) jsonrpc2.Message {
+	msg, err := jsonrpc2.NewCall(newID(id), method, params)
+	if err != nil {
+		panic(err)
+	}
+	return msg
+}
+
+func newResponse(id interface{}, result interface{}, rerr error) jsonrpc2.Message {
+	msg, err := jsonrpc2.NewResponse(newID(id), result, rerr)
+	if err != nil {
+		panic(err)
+	}
+	return msg
+}
+
+func checkJSON(t *testing.T, got, want []byte) {
+	// compare the compact form, to allow for formatting differences
+	g := &bytes.Buffer{}
+	if err := json.Compact(g, []byte(got)); err != nil {
+		t.Fatal(err)
+	}
+	w := &bytes.Buffer{}
+	if err := json.Compact(w, []byte(want)); err != nil {
+		t.Fatal(err)
+	}
+	if g.String() != w.String() {
+		t.Errorf("encoded message does not match\nGot:\n%s\nWant:\n%s", g, w)
+	}
+}