Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
package http
import (
"bytes"
"context"
"encoding/base64"
"net/http"
"net/url"
"strings"
"gfx.cafe/open/jrpc/pkg/jsonrpc"
"gfx.cafe/open/jrpc/pkg/serverutil"
"github.com/gfx-labs/sse"
)
var _ jsonrpc.ReaderWriter = (*SseCodec)(nil)
// SseCodec is used for subscriptions over http.
// note that every message is buffered multiple times - it's a bit inefficient.
// if you need more efficient streaming of large blobs, consider using a different interface
type SseCodec struct {
ctx context.Context
cn context.CancelFunc
r *http.Request
w http.ResponseWriter
i jsonrpc.PeerInfo
sink *sse.EventSink
f http.Flusher
msgs *serverutil.Bundle
cur bytes.Buffer
}
func NewSseCodec(w http.ResponseWriter, r *http.Request) (*SseCodec, error) {
// attempt to upgrade
c := &SseCodec{
r: r,
w: w,
i: jsonrpc.PeerInfo{
Transport: "http",
RemoteAddr: r.RemoteAddr,
HTTP: r.Clone(r.Context()),
},
}
c.ctx, c.cn = context.WithCancel(r.Context())
eventSink, err := sse.DefaultUpgrader.Upgrade(w, r)
if err != nil {
return nil, err
}
c.sink = eventSink
method_up := r.URL.Query().Get("method")
if method_up == "" {
method_up = strings.TrimPrefix(r.URL.Path, "/")
}
params, _ := url.QueryUnescape(r.URL.Query().Get("params"))
var param []byte
// try to read params as base64
if pb, err := base64.URLEncoding.DecodeString(params); err == nil {
param = pb
} else {
// otherwise just take them raw
param = []byte(params)
}
id := r.URL.Query().Get("id")
if id == "" {
id = "1"
}
c.msgs = &serverutil.Bundle{
Messages: []*jsonrpc.Message{{
ID: jsonrpc.NewId(id),
Method: method_up,
Params: param,
}},
Batch: false,
}
return c, nil
}
// gets the peer info
func (c *SseCodec) PeerInfo() jsonrpc.PeerInfo {
return c.i
}
func (c *SseCodec) ReadBatch(ctx context.Context) ([]*jsonrpc.Message, bool, error) {
if c.msgs == nil {
return nil, false, context.Canceled
}
defer func() {
c.msgs = nil
}()
return c.msgs.Messages, c.msgs.Batch, nil
}
// closes the connection
func (c *SseCodec) Write(p []byte) (n int, err error) {
return c.cur.Write(p)
}
func (c *SseCodec) Flush() error {
c.sink.Encode(&sse.Event{
Event: []byte("object"),
Data: &c.cur,
})
c.w.Write([]byte{'\n'})
if c.f != nil {
c.f.Flush()
}
return nil
}
func (c *SseCodec) Close() error {
if c.f != nil {
c.f.Flush()
}
c.cn()
return nil
}
// Closed returns a channel which is closed when the connection is closed.
func (c *SseCodec) Closed() <-chan struct{} {
return c.ctx.Done()
}
// RemoteAddr returns the peer address of the connection.
func (c *SseCodec) RemoteAddr() string {
return c.r.RemoteAddr
}