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
130
131
132
133
134
135
136
137
138
139
|
package main
import (
"context"
"encoding/json"
"fmt"
"net/http"
"sync/atomic"
"muhq.space/go/wrms/llog"
"github.com/google/uuid"
)
type Connection struct {
wrms *Wrms
Id uuid.UUID
nr uint64
closing atomic.Bool
refs atomic.Int64
Events chan Event
w http.ResponseWriter
flusher http.Flusher
ctx context.Context
cancel func()
nextEvent uint64
}
const EVENT_BUFFER_SIZE = 3
func (wrms *Wrms) newConnection(id uuid.UUID,
w http.ResponseWriter,
ctx context.Context,
cncl func()) *Connection {
// prepare the flusher
flusher, ok := w.(http.Flusher)
if !ok {
llog.Fatal("Flusher type assertion failed")
}
conn := &Connection{
wrms: wrms,
Id: id,
nr: wrms.nextConnNr.Add(1),
Events: make(chan Event, EVENT_BUFFER_SIZE),
w: w,
flusher: flusher,
ctx: ctx,
cancel: cncl,
}
wrms.addConn(conn)
return conn
}
func (c *Connection) Send(ev Event) {
llog.DDebug("Sending %v to %v", ev, c.Id)
// Connection is closing -> not write to it
if c.closing.Load() {
return
}
// Register us as sender
c.refs.Add(1)
// Send the event
c.Events <- ev
// Deregister us as sender
c.refs.Add(-1)
}
func (conn *Connection) _send(evs []interface{}) {
for _, ev := range evs {
data, err := json.Marshal(ev)
if err != nil {
llog.Fatal("Encoding the initial command %v failed with %s", ev, err)
}
sdata := string(data)
llog.Debug("Sending ev %s to %s", sdata, conn.Id)
fmt.Fprintf(conn.w, "data: %s\n\n", sdata)
conn.flusher.Flush()
}
}
func (conn *Connection) _sendEv(ev Event) {
conn._send([]interface{}{ev})
}
func (conn *Connection) serve() {
// Map to store future events to send them in order
toSend := make(map[uint64]*Event)
for {
var ev Event
// We have the next event -> send it
if toSend[conn.nextEvent] != nil {
ev = *toSend[conn.nextEvent]
delete(toSend, conn.nextEvent)
// We have not received the next event yet -> receive a new event
} else {
llog.DDebug("%v: Awaiting event %d", conn.Id, conn.nextEvent)
select {
case ev = <-conn.Events:
case <-conn.ctx.Done():
return
}
// We received a future object -> store it and continue
if ev.Id > conn.nextEvent {
llog.DDebug("%v: Received future event %d", conn.Id, ev.Id)
toSend[ev.Id] = &ev
continue
}
}
// increment the expected event id
if ev.Id == conn.nextEvent {
conn.nextEvent++
}
conn._sendEv(ev)
}
}
func (c *Connection) Close() {
llog.Info("Closing connection %s", c.Id)
// Remove the closing connection from the map
wrms.delConn(c)
// Announce that the connection is going to be closed
c.closing.Store(true)
// Consume all events from the registered senders
for c.refs.Load() > 0 {
select {
case <-c.Events:
default:
}
}
close(c.Events)
}
|
