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
|
package comm
import (
"errors"
"log"
"github.com/francoispqt/gojay"
"golang.org/x/net/websocket"
)
// A basic message for our WebSocket app
type Message struct {
Message string
UserName string
}
func (m *Message) UnmarshalJSONObject(dec *gojay.Decoder, k string) error {
switch k {
case "message":
return dec.AddString(&m.Message)
case "userName":
return dec.AddString(&m.UserName)
}
return nil
}
func (m *Message) NKeys() int {
return 2
}
func (m *Message) MarshalJSONObject(enc *gojay.Encoder) {
enc.AddStringKey("message", m.Message)
enc.AddStringKey("userName", m.UserName)
}
func (u *Message) IsNil() bool {
return u == nil
}
// Here are defined our communication types
type Sender chan gojay.MarshalerJSONObject
func (s Sender) MarshalStream(enc *gojay.StreamEncoder) {
select {
case <-enc.Done():
return
case m := <-s:
enc.AddObject(m)
}
}
type Receiver chan *Message
func (s Receiver) UnmarshalStream(dec *gojay.StreamDecoder) error {
m := &Message{}
if err := dec.AddObject(m); err != nil {
return err
}
s <- m
return nil
}
type SenderReceiver struct {
Send Sender
Receive Receiver
Dec *gojay.StreamDecoder
Enc *gojay.StreamEncoder
Conn *websocket.Conn
}
func (sc *SenderReceiver) SetReceiver() {
sc.Receive = Receiver(make(chan *Message))
sc.Dec = gojay.Stream.BorrowDecoder(sc.Conn)
go sc.Dec.DecodeStream(sc.Receive)
}
func (sc *SenderReceiver) SetSender(nCons int) {
sc.Send = Sender(make(chan gojay.MarshalerJSONObject))
sc.Enc = gojay.Stream.BorrowEncoder(sc.Conn).NConsumer(nCons).LineDelimited()
go sc.Enc.EncodeStream(sc.Send)
}
func (sc *SenderReceiver) SendMessage(m gojay.MarshalerJSONObject) error {
select {
case <-sc.Enc.Done():
return errors.New("sender closed")
case sc.Send <- m:
log.Print("message sent by client: ", m)
return nil
}
}
func (c *SenderReceiver) OnMessage(f func(*Message)) error {
for {
select {
case <-c.Dec.Done():
return errors.New("receiver closed")
case m := <-c.Receive:
f(m)
}
}
}
func (sc *SenderReceiver) Init(sender int) *SenderReceiver {
sc.SetSender(sender)
sc.SetReceiver()
return sc
}
|
