JSON-RPC 2.0 over websockets
Initial Commit
+531
+11
.gitignore
+11
.gitignore
+21
LICENSE
+21
LICENSE
···
1
+
MIT License
2
+
3
+
Copyright (c) 2026 Josh Ghiloni
4
+
5
+
Permission is hereby granted, free of charge, to any person obtaining a copy
6
+
of this software and associated documentation files (the "Software"), to deal
7
+
in the Software without restriction, including without limitation the rights
8
+
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9
+
copies of the Software, and to permit persons to whom the Software is
10
+
furnished to do so, subject to the following conditions:
11
+
12
+
The above copyright notice and this permission notice shall be included in all
13
+
copies or substantial portions of the Software.
14
+
15
+
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16
+
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17
+
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18
+
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19
+
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20
+
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
21
+
SOFTWARE.
+331
client.go
+331
client.go
···
1
+
package jsonrpc
2
+
3
+
import (
4
+
"context"
5
+
"crypto/x509"
6
+
"encoding/pem"
7
+
"errors"
8
+
"fmt"
9
+
"log/slog"
10
+
"net/http"
11
+
"strconv"
12
+
"sync"
13
+
"sync/atomic"
14
+
"time"
15
+
16
+
"github.com/coder/websocket"
17
+
"github.com/coder/websocket/wsjson"
18
+
"github.com/goccy/go-json"
19
+
)
20
+
21
+
// WebsocketClient allows users to connect to a websocket that serves a JSON-RPC
22
+
// 2.0 API and send method calls, receive responses, and subscribe to notifications
23
+
// It is safe to use a single [WebsocketClient] across multiple goroutines.
24
+
//
25
+
// Because of the non-trivial overhead associated with connecting to websockets
26
+
// and typically the overahead of authenticating a session, it's recommended to
27
+
// keep a session alive for as long as possible, though the details of that will
28
+
// vary from provider to provider
29
+
type WebsocketClient struct {
30
+
conn *websocket.Conn
31
+
subscribers map[string][]NotificationSubscription
32
+
subscriberMapMu sync.RWMutex
33
+
activeRequests map[ID]chan Response
34
+
reqMapMu sync.Mutex
35
+
lastID atomic.Pointer[ID]
36
+
nextID func(ID) ID
37
+
idGenMu sync.Mutex
38
+
logger *slog.Logger
39
+
active atomic.Bool
40
+
}
41
+
42
+
// ClientOptions are optional parameters to send to [NewClient] that have reasonable
43
+
// defaults if omitted
44
+
type ClientOptions struct {
45
+
// SkipTLSValidation tells clients connecting to wss:// URLs to not verify
46
+
// TLS certs, allowing it to connect to servers without trusted certificates.
47
+
// If possible, opt for CACert instead
48
+
SkipTLSValidation bool
49
+
50
+
// CACert adds provided Certificate to the underlying HTTP Client's
51
+
// trusted certificate pool. It must be a []byte, string, or [*crypto/x509.Certificate].
52
+
// If it is a []byte or string, it must be a PEM-encoded x509 string.
53
+
CACert any
54
+
55
+
// Logger adds a specific [*log/slog.Logger] to the client. If not provided,
56
+
// [log/slog.Default] will be used
57
+
Logger *slog.Logger
58
+
59
+
// HTTPClient allows a user to use a specific underlying [*net/http.Client]
60
+
// to be used when making the websocket connection. If this is set, the values
61
+
// of SkipTLSValidation and CACert are ignored. If it is not set, [net/http.DefaultClient]
62
+
// is used
63
+
HTTPClient *http.Client
64
+
65
+
// IDGenerator allows the user to specify a func to take the last ID known
66
+
// to the collection and generate the next. Because JSON-RPC correlates
67
+
// requests and responses with IDs, it is imperative these are unique within
68
+
// a connection, or unexpected behavior may occur. If this is omitted, the
69
+
// function will respond with the current timestamp with nanosecond precision,
70
+
// as a base-32 integer (as opposed to being base-32 encoded)
71
+
IDGenerator func(lastID ID) ID
72
+
73
+
// DialOptions contain further options to send to the underlying websocket
74
+
// connection. Note that these will be passed as-is, with the exception being
75
+
// HTTPClient, which will be overridden with the value of HTTPClient generated
76
+
// in NewClient
77
+
DialOptions *websocket.DialOptions
78
+
}
79
+
80
+
func defaultIDGen(_ ID) ID {
81
+
return ID(strconv.FormatInt(time.Now().UTC().UnixNano(), 32))
82
+
}
83
+
84
+
// NewClient attempts to connect to the given URL (which must have a ws:// or
85
+
// wss:// scheme) with the given options. If successful, it returns a client
86
+
// with an open connection but is not yet listening. For that, you must call
87
+
// [*WebsocketClient.Start]
88
+
func NewClient(ctx context.Context, serverURL string, options *ClientOptions) (*WebsocketClient, error) {
89
+
var (
90
+
err error
91
+
resp *http.Response
92
+
)
93
+
94
+
nextID := defaultIDGen
95
+
if options.IDGenerator != nil {
96
+
nextID = options.IDGenerator
97
+
}
98
+
c := &WebsocketClient{
99
+
subscribers: make(map[string][]NotificationSubscription),
100
+
activeRequests: make(map[ID]chan Response),
101
+
nextID: nextID,
102
+
logger: options.Logger,
103
+
}
104
+
105
+
if c.logger == nil {
106
+
c.logger = slog.Default()
107
+
}
108
+
109
+
httpClient, err := getHTTPClient(options)
110
+
if err != nil {
111
+
return nil, err
112
+
}
113
+
114
+
if options.DialOptions == nil {
115
+
options.DialOptions = new(websocket.DialOptions)
116
+
}
117
+
118
+
options.DialOptions.HTTPClient = httpClient
119
+
120
+
c.conn, resp, err = websocket.Dial(ctx, serverURL, options.DialOptions)
121
+
if resp != nil && resp.Body != nil {
122
+
resp.Body.Close()
123
+
}
124
+
return c, err
125
+
}
126
+
127
+
// Start generates the first ID for the server, and starts listening for messages
128
+
func (w *WebsocketClient) Start(ctx context.Context) {
129
+
// set the first ID
130
+
w.getNextID()
131
+
go w.listen(ctx)
132
+
}
133
+
134
+
// Close implements [io.Closer]. It closes any active response channels and deletes
135
+
// all notification subscriptions, as well as closing the websocket connection
136
+
// without waiting for a response
137
+
func (w *WebsocketClient) Close() error {
138
+
safeCloseChan := func(c chan Response) {
139
+
defer func() {
140
+
// closing a closed channel panics, but we just want to ignore that
141
+
recover()
142
+
}()
143
+
close(c)
144
+
}
145
+
146
+
w.reqMapMu.Lock()
147
+
for _, r := range w.activeRequests {
148
+
safeCloseChan(r)
149
+
}
150
+
w.reqMapMu.Unlock()
151
+
152
+
w.subscriberMapMu.Lock()
153
+
for m := range w.subscribers {
154
+
delete(w.subscribers, m)
155
+
}
156
+
157
+
return w.conn.CloseNow()
158
+
}
159
+
160
+
func (w *WebsocketClient) Call(ctx context.Context, method string, params ...any) (response chan Response, err error) {
161
+
select {
162
+
case <-ctx.Done():
163
+
return nil, ctx.Err()
164
+
default:
165
+
}
166
+
167
+
request := Request{
168
+
Notification: Notification{
169
+
Method: method,
170
+
},
171
+
ID: w.getNextID(),
172
+
}
173
+
174
+
request.Params, err = json.Marshal(params)
175
+
if err != nil {
176
+
return
177
+
}
178
+
179
+
response = make(chan Response, 1)
180
+
w.reqMapMu.Lock()
181
+
w.activeRequests[request.ID] = response
182
+
w.reqMapMu.Unlock()
183
+
184
+
return response, wsjson.Write(ctx, w.conn, request)
185
+
}
186
+
187
+
func (w *WebsocketClient) CallSynchronous(ctx context.Context, method string, params ...any) (any, error) {
188
+
select {
189
+
case <-ctx.Done():
190
+
return nil, ctx.Err()
191
+
default:
192
+
}
193
+
194
+
response, err := w.Call(ctx, method, params...)
195
+
if err != nil {
196
+
return nil, err
197
+
}
198
+
199
+
select {
200
+
case <-ctx.Done():
201
+
return nil, ctx.Err()
202
+
case r := <-response:
203
+
var a any
204
+
err = r.Result(&a)
205
+
return a, errors.Join(r.Error(), err)
206
+
}
207
+
}
208
+
209
+
func (w *WebsocketClient) listen(ctx context.Context) {
210
+
for {
211
+
select {
212
+
case <-ctx.Done():
213
+
return
214
+
default:
215
+
}
216
+
217
+
w.active.Store(true)
218
+
219
+
var e envelope
220
+
err := wsjson.Read(ctx, w.conn, &e)
221
+
if err != nil {
222
+
w.logger.Error("error reading message from websocket", "error", err)
223
+
continue
224
+
}
225
+
226
+
// first try to unmarshal it as a response
227
+
var response apiResponse
228
+
if err = e.unwrap(&response); err == nil {
229
+
w.reqMapMu.Lock()
230
+
r, ok := w.activeRequests[response.ID()]
231
+
if !ok {
232
+
w.logger.Warn("message received for unknown request ID", "id", response.ID())
233
+
w.reqMapMu.Unlock()
234
+
continue
235
+
}
236
+
delete(w.activeRequests, response.ID())
237
+
w.reqMapMu.Unlock()
238
+
239
+
r <- response
240
+
close(r)
241
+
continue
242
+
}
243
+
244
+
var notif Notification
245
+
if err = e.unwrap(¬if); err == nil {
246
+
w.subscriberMapMu.RLock()
247
+
listeners := w.subscribers[notif.Method]
248
+
w.subscriberMapMu.RUnlock()
249
+
250
+
for _, listener := range listeners {
251
+
listener(notif)
252
+
}
253
+
}
254
+
}
255
+
}
256
+
257
+
func (w *WebsocketClient) IsActive() bool {
258
+
return w.active.Load()
259
+
}
260
+
261
+
func (w *WebsocketClient) SubscribeToNotification(name string, listener NotificationSubscription) {
262
+
w.subscriberMapMu.Lock()
263
+
defer w.subscriberMapMu.Unlock()
264
+
265
+
w.subscribers[name] = append(w.subscribers[name], listener)
266
+
}
267
+
268
+
func (w *WebsocketClient) getNextID() ID {
269
+
w.idGenMu.Lock()
270
+
next := new(w.nextID(""))
271
+
w.lastID.Store(next)
272
+
w.idGenMu.Unlock()
273
+
274
+
return *next
275
+
}
276
+
277
+
func getHTTPClient(options *ClientOptions) (*http.Client, error) {
278
+
// if the client is explicitly set, we're done. ignore skiptlsvalidation
279
+
if options.HTTPClient != nil {
280
+
return options.HTTPClient, nil
281
+
}
282
+
283
+
client := http.DefaultClient
284
+
tr := client.Transport.(*http.Transport)
285
+
var err error
286
+
switch {
287
+
case options.CACert != nil:
288
+
var cert *x509.Certificate
289
+
switch c := options.CACert.(type) {
290
+
case string:
291
+
cert, err = getCertFromBytes([]byte(c))
292
+
if err != nil {
293
+
return nil, err
294
+
}
295
+
case []byte:
296
+
cert, err = getCertFromBytes(c)
297
+
if err != nil {
298
+
return nil, err
299
+
}
300
+
case *x509.Certificate:
301
+
cert = c
302
+
default:
303
+
return nil, fmt.Errorf("expected CACert to be []byte, string, or *x509.Certificate, got %T", options.CACert)
304
+
}
305
+
306
+
certPool := tr.TLSClientConfig.RootCAs
307
+
if certPool == nil {
308
+
certPool, err = x509.SystemCertPool()
309
+
if err != nil {
310
+
return nil, err
311
+
}
312
+
}
313
+
certPool.AddCert(cert)
314
+
315
+
// TODO check if this is required with debugging
316
+
tr.TLSClientConfig.RootCAs = certPool
317
+
case options.SkipTLSValidation:
318
+
tr.TLSClientConfig.InsecureSkipVerify = true
319
+
}
320
+
321
+
return client, nil
322
+
}
323
+
324
+
func getCertFromBytes(b []byte) (*x509.Certificate, error) {
325
+
block, _ := pem.Decode(b)
326
+
if block != nil {
327
+
return x509.ParseCertificate(block.Bytes)
328
+
}
329
+
330
+
return nil, errors.New("bytes contained no PEM data")
331
+
}
+24
go.mod
+24
go.mod
···
1
+
module tangled.org/joshghiloni.me/jsonrpc-ws
2
+
3
+
go 1.26.1
4
+
5
+
require (
6
+
github.com/Masterminds/semver/v3 v3.4.0 // indirect
7
+
github.com/coder/websocket v1.8.14 // indirect
8
+
github.com/go-logr/logr v1.4.3 // indirect
9
+
github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
10
+
github.com/goccy/go-json v0.10.6 // indirect
11
+
github.com/google/go-cmp v0.7.0 // indirect
12
+
github.com/google/pprof v0.0.0-20260115054156-294ebfa9ad83 // indirect
13
+
github.com/onsi/ginkgo/v2 v2.28.1 // indirect
14
+
github.com/onsi/gomega v1.39.1 // indirect
15
+
go.yaml.in/yaml/v3 v3.0.4 // indirect
16
+
golang.org/x/mod v0.32.0 // indirect
17
+
golang.org/x/net v0.49.0 // indirect
18
+
golang.org/x/sync v0.19.0 // indirect
19
+
golang.org/x/sys v0.40.0 // indirect
20
+
golang.org/x/text v0.33.0 // indirect
21
+
golang.org/x/tools v0.41.0 // indirect
22
+
)
23
+
24
+
tool github.com/onsi/ginkgo/v2/ginkgo
+33
go.sum
+33
go.sum
···
1
+
github.com/Masterminds/semver/v3 v3.4.0 h1:Zog+i5UMtVoCU8oKka5P7i9q9HgrJeGzI9SA1Xbatp0=
2
+
github.com/Masterminds/semver/v3 v3.4.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM=
3
+
github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g=
4
+
github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
5
+
github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
6
+
github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
7
+
github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI=
8
+
github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8=
9
+
github.com/goccy/go-json v0.10.6 h1:p8HrPJzOakx/mn/bQtjgNjdTcN+/S6FcG2CTtQOrHVU=
10
+
github.com/goccy/go-json v0.10.6/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
11
+
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
12
+
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
13
+
github.com/google/pprof v0.0.0-20260115054156-294ebfa9ad83 h1:z2ogiKUYzX5Is6zr/vP9vJGqPwcdqsWjOt+V8J7+bTc=
14
+
github.com/google/pprof v0.0.0-20260115054156-294ebfa9ad83/go.mod h1:MxpfABSjhmINe3F1It9d+8exIHFvUqtLIRCdOGNXqiI=
15
+
github.com/onsi/ginkgo/v2 v2.28.1 h1:S4hj+HbZp40fNKuLUQOYLDgZLwNUVn19N3Atb98NCyI=
16
+
github.com/onsi/ginkgo/v2 v2.28.1/go.mod h1:CLtbVInNckU3/+gC8LzkGUb9oF+e8W8TdUsxPwvdOgE=
17
+
github.com/onsi/gomega v1.39.1 h1:1IJLAad4zjPn2PsnhH70V4DKRFlrCzGBNrNaru+Vf28=
18
+
github.com/onsi/gomega v1.39.1/go.mod h1:hL6yVALoTOxeWudERyfppUcZXjMwIMLnuSfruD2lcfg=
19
+
go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
20
+
go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
21
+
golang.org/x/mod v0.32.0 h1:9F4d3PHLljb6x//jOyokMv3eX+YDeepZSEo3mFJy93c=
22
+
golang.org/x/mod v0.32.0/go.mod h1:SgipZ/3h2Ci89DlEtEXWUk/HteuRin+HHhN+WbNhguU=
23
+
golang.org/x/net v0.49.0 h1:eeHFmOGUTtaaPSGNmjBKpbng9MulQsJURQUAfUwY++o=
24
+
golang.org/x/net v0.49.0/go.mod h1:/ysNB2EvaqvesRkuLAyjI1ycPZlQHM3q01F02UY/MV8=
25
+
golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4=
26
+
golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
27
+
golang.org/x/sys v0.40.0 h1:DBZZqJ2Rkml6QMQsZywtnjnnGvHza6BTfYFWY9kjEWQ=
28
+
golang.org/x/sys v0.40.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
29
+
golang.org/x/text v0.33.0 h1:B3njUFyqtHDUI5jMn1YIr5B0IE2U0qck04r6d4KPAxE=
30
+
golang.org/x/text v0.33.0/go.mod h1:LuMebE6+rBincTi9+xWTY8TztLzKHc/9C1uBCG27+q8=
31
+
golang.org/x/tools v0.41.0 h1:a9b8iMweWG+S0OBnlU36rzLp20z1Rp10w+IY2czHTQc=
32
+
golang.org/x/tools v0.41.0/go.mod h1:XSY6eDqxVNiYgezAVqqCeihT4j1U2CCsqvH3WhQpnlg=
33
+
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+111
types.go
+111
types.go
···
1
+
package jsonrpc
2
+
3
+
import (
4
+
"fmt"
5
+
6
+
"github.com/goccy/go-json"
7
+
)
8
+
9
+
const (
10
+
jrVal = "2.0"
11
+
)
12
+
13
+
type envelope struct {
14
+
E_ string `json:"jsonrpc"`
15
+
Val json.RawMessage `json:",inline"`
16
+
}
17
+
18
+
func wrap(v any) (*envelope, error) {
19
+
msg, err := json.Marshal(v)
20
+
if err != nil {
21
+
return nil, err
22
+
}
23
+
return &envelope{E_: jrVal, Val: msg}, nil
24
+
}
25
+
26
+
func (e *envelope) unwrap(target any) error {
27
+
return json.Unmarshal(e.Val, target)
28
+
}
29
+
30
+
type ID string
31
+
32
+
func (i *ID) UnmarshalJSON(data []byte) error {
33
+
var n json.Number
34
+
if err := json.Unmarshal(data, &n); err != nil {
35
+
return err
36
+
}
37
+
38
+
_, ierr := n.Int64()
39
+
_, ferr := n.Float64()
40
+
41
+
if ierr != nil && ferr == nil {
42
+
return fmt.Errorf("if ID is a number it must be a whole number")
43
+
}
44
+
45
+
*i = ID(n)
46
+
return nil
47
+
}
48
+
49
+
type Notification struct {
50
+
Method string `json:"method"`
51
+
Params json.RawMessage `json:"params,omitempty"`
52
+
}
53
+
54
+
type Request struct {
55
+
Notification `json:",inline"`
56
+
ID ID `json:"id"`
57
+
}
58
+
59
+
type ServerError struct {
60
+
Code int `json:"code"`
61
+
Message string `json:"message"`
62
+
RawData json.RawMessage `json:"data,omitempty"`
63
+
cause error
64
+
}
65
+
66
+
func (e *ServerError) Error() string {
67
+
return e.Message
68
+
}
69
+
70
+
func (e *ServerError) InspectData(target any) error {
71
+
if err := json.Unmarshal(e.RawData, target); err != nil {
72
+
return err
73
+
}
74
+
75
+
te, ok := target.(error)
76
+
if ok {
77
+
e.cause = te
78
+
}
79
+
80
+
return nil
81
+
}
82
+
83
+
func (e *ServerError) Unwrap() error {
84
+
return e.cause
85
+
}
86
+
87
+
type Response interface {
88
+
ID() ID
89
+
Result(target any) error
90
+
Error() error
91
+
}
92
+
93
+
type apiResponse struct {
94
+
ID_ ID `json:"id"`
95
+
Result_ json.RawMessage `json:"result,omitempty"`
96
+
Err *ServerError `json:"error,omitempty"`
97
+
}
98
+
99
+
func (a apiResponse) ID() ID {
100
+
return a.ID_
101
+
}
102
+
103
+
func (a apiResponse) Result(target any) error {
104
+
return json.Unmarshal(a.Result_, target)
105
+
}
106
+
107
+
func (a apiResponse) Error() error {
108
+
return a.Err
109
+
}
110
+
111
+
type NotificationSubscription func(Notification)