~fnux/yggdrasil-go-coap

ref: e042d7baf5169b5c24e7261d57260a87506cb2db yggdrasil-go-coap/client.go -rw-r--r-- 13.3 KiB
e042d7ba — Jozef Kralik fix sending message when peerMaxMessageSize is not set 2 years ago
                                                                                
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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
package coap

// A client implementation.

import (
	"context"
	"crypto/tls"
	"io"
	"net"
	"strings"
	"time"

	kitNet "github.com/go-ocf/kit/net"
)

// A ClientConn represents a connection to a COAP server.
type ClientConn struct {
	srv          *Server
	client       *Client
	commander    *ClientCommander
	shutdownSync chan error
	multicast    bool
}

// A Client defines parameters for a COAP client.
type Client struct {
	Net            string        // if "tcp" or "tcp-tls" (COAP over TLS) a TCP query will be initiated, otherwise an UDP one (default is "" for UDP) or "udp-mcast" for multicast
	MaxMessageSize uint32        // Max message size that could be received from peer. If not set it defaults to 1152 B.
	TLSConfig      *tls.Config   // TLS connection configuration
	DialTimeout    time.Duration // set Timeout for dialer
	ReadTimeout    time.Duration // net.ClientConn.SetReadTimeout value for connections, defaults to 1 hour - overridden by Timeout when that value is non-zero
	WriteTimeout   time.Duration // net.ClientConn.SetWriteTimeout value for connections, defaults to 1 hour - overridden by Timeout when that value is non-zero
	HeartBeat      time.Duration // Defines wake up interval from operations Read, Write over connection. defaults is 100ms.

	Handler              HandlerFunc     // default handler for handling messages from server
	NotifySessionEndFunc func(err error) // if NotifySessionEndFunc is set it is called when TCP/UDP session was ended.

	BlockWiseTransfer    *bool         // Use blockWise transfer for transfer payload (default for UDP it's enabled, for TCP it's disable)
	BlockWiseTransferSzx *BlockWiseSzx // Set maximal block size of payload that will be send in fragment

	DisableTCPSignalMessages bool // Disable tcp signal messages
}

func (c *Client) readTimeout() time.Duration {
	if c.ReadTimeout != 0 {
		return c.ReadTimeout
	}
	return coapTimeout
}

func (c *Client) writeTimeout() time.Duration {
	if c.WriteTimeout != 0 {
		return c.WriteTimeout
	}
	return coapTimeout
}

func listenUDP(network, address string) (*net.UDPAddr, *net.UDPConn, error) {
	var a *net.UDPAddr
	var err error
	if a, err = net.ResolveUDPAddr(network, address); err != nil {
		return nil, nil, err
	}
	var udpConn *net.UDPConn
	if udpConn, err = net.ListenUDP(network, a); err != nil {
		return nil, nil, err
	}
	if err := kitNet.SetUDPSocketOptions(udpConn); err != nil {
		return nil, nil, err
	}
	return a, udpConn, nil
}

func (c *Client) Dial(address string) (clientConn *ClientConn, err error) {
	return c.DialWithContext(context.Background(), address)
}

// DialContext connects to the address on the named network.
func (c *Client) DialWithContext(ctx context.Context, address string) (clientConn *ClientConn, err error) {

	var conn net.Conn
	var network string
	var sessionUPDData *kitNet.ConnUDPContext

	dialer := &net.Dialer{Timeout: c.DialTimeout}
	BlockWiseTransfer := false
	BlockWiseTransferSzx := BlockWiseSzx1024
	multicast := false

	switch c.Net {
	case "tcp-tls", "tcp4-tls", "tcp6-tls":
		network = strings.TrimSuffix(c.Net, "-tls")
		conn, err = tls.DialWithDialer(dialer, network, address, c.TLSConfig)
		if err != nil {
			return nil, err
		}
		BlockWiseTransferSzx = BlockWiseSzxBERT
	case "tcp", "tcp4", "tcp6":
		network = c.Net
		conn, err = dialer.DialContext(ctx, c.Net, address)
		if err != nil {
			return nil, err
		}
		BlockWiseTransferSzx = BlockWiseSzxBERT
	case "udp", "udp4", "udp6", "":
		network = c.Net
		if network == "" {
			network = "udp"
		}
		if conn, err = dialer.DialContext(ctx, network, address); err != nil {
			return nil, err
		}
		sessionUPDData = kitNet.NewConnUDPWithContext(conn.(*net.UDPConn).RemoteAddr().(*net.UDPAddr), nil)
		BlockWiseTransfer = true
	case "udp-mcast", "udp4-mcast", "udp6-mcast":
		network = strings.TrimSuffix(c.Net, "-mcast")
		a, udpConn, err := listenUDP(network, address)
		if err != nil {
			return nil, err
		}
		sessionUPDData = kitNet.NewConnUDPWithContext(a, nil)
		conn = udpConn
		BlockWiseTransfer = true
		multicast = true
	default:
		return nil, ErrInvalidNetParameter
	}

	if c.BlockWiseTransfer != nil {
		BlockWiseTransfer = *c.BlockWiseTransfer
	}

	if c.BlockWiseTransferSzx != nil {
		BlockWiseTransferSzx = *c.BlockWiseTransferSzx
	}

	//sync := make(chan bool)
	clientConn = &ClientConn{
		srv: &Server{
			Net:                      network,
			TLSConfig:                c.TLSConfig,
			Conn:                     conn,
			ReadTimeout:              c.readTimeout(),
			WriteTimeout:             c.writeTimeout(),
			MaxMessageSize:           c.MaxMessageSize,
			BlockWiseTransfer:        &BlockWiseTransfer,
			BlockWiseTransferSzx:     &BlockWiseTransferSzx,
			DisableTCPSignalMessages: c.DisableTCPSignalMessages,
			NotifySessionEndFunc: func(s *ClientConn, err error) {
				if c.NotifySessionEndFunc != nil {
					c.NotifySessionEndFunc(err)
				}
			},
			newSessionTCPFunc: func(connection *kitNet.Conn, srv *Server) (networkSession, error) {
				return clientConn.commander.networkSession, nil
			},
			newSessionUDPFunc: func(connection *kitNet.ConnUDP, srv *Server, sessionUDPData *kitNet.ConnUDPContext) (networkSession, error) {
				if sessionUDPData.RemoteAddr().String() == clientConn.commander.networkSession.RemoteAddr().String() {
					if s, ok := clientConn.commander.networkSession.(*blockWiseSession); ok {
						s.networkSession.(*sessionUDP).sessionUDPData = sessionUDPData
					} else {
						clientConn.commander.networkSession.(*sessionUDP).sessionUDPData = sessionUDPData
					}
					return clientConn.commander.networkSession, nil
				}
				session, err := newSessionUDP(connection, srv, sessionUDPData)
				if err != nil {
					return nil, err
				}
				if session.blockWiseEnabled() {
					return &blockWiseSession{networkSession: session}, nil
				}
				return session, nil
			},
			Handler: c.Handler,
		},
		shutdownSync: make(chan error, 1),
		multicast:    multicast,
		commander:    &ClientCommander{},
	}

	switch clientConn.srv.Conn.(type) {
	case *net.TCPConn, *tls.Conn:
		session, err := newSessionTCP(kitNet.NewConn(clientConn.srv.Conn, clientConn.srv.heartBeat()), clientConn.srv)
		if err != nil {
			return nil, err
		}
		if session.blockWiseEnabled() {
			clientConn.commander.networkSession = &blockWiseSession{networkSession: session}
		} else {
			clientConn.commander.networkSession = session
		}
	case *net.UDPConn:
		// WriteContextMsgUDP returns error when addr is filled in SessionUDPData for connected socket
		kitNet.SetUDPSocketOptions(clientConn.srv.Conn.(*net.UDPConn))
		session, err := newSessionUDP(kitNet.NewConnUDP(clientConn.srv.Conn.(*net.UDPConn), clientConn.srv.heartBeat()), clientConn.srv, sessionUPDData)
		if err != nil {
			return nil, err
		}
		if session.blockWiseEnabled() {
			clientConn.commander.networkSession = &blockWiseSession{networkSession: session}
		} else {
			clientConn.commander.networkSession = session
		}
	}

	go func() {
		err := clientConn.srv.ActivateAndServe()
		select {
		case clientConn.shutdownSync <- err:
		}
	}()
	clientConn.client = c

	return clientConn, nil
}

func (co *ClientConn) networkSession() networkSession {
	return co.commander.networkSession
}

// LocalAddr implements the networkSession.LocalAddr method.
func (co *ClientConn) LocalAddr() net.Addr {
	return co.commander.LocalAddr()
}

// RemoteAddr implements the networkSession.RemoteAddr method.
func (co *ClientConn) RemoteAddr() net.Addr {
	return co.commander.RemoteAddr()
}

func (co *ClientConn) Exchange(m Message) (Message, error) {
	return co.commander.ExchangeWithContext(context.Background(), m)
}

// ExchangeContext performs a synchronous query. It sends the message m to the address
// contained in a and waits for a reply.
//
// ExchangeContext does not retry a failed query, nor will it fall back to TCP in
// case of truncation.
// To specify a local address or a timeout, the caller has to set the `Client.Dialer`
// attribute appropriately
func (co *ClientConn) ExchangeWithContext(ctx context.Context, m Message) (Message, error) {
	if co.multicast {
		return nil, ErrNotSupported
	}
	return co.commander.ExchangeWithContext(ctx, m)
}

// NewMessage Create message for request
func (co *ClientConn) NewMessage(p MessageParams) Message {
	return co.commander.NewMessage(p)
}

// NewGetRequest creates get request
func (co *ClientConn) NewGetRequest(path string) (Message, error) {
	return co.commander.NewGetRequest(path)
}

// NewPostRequest creates post request
func (co *ClientConn) NewPostRequest(path string, contentFormat MediaType, body io.Reader) (Message, error) {
	return co.commander.NewPostRequest(path, contentFormat, body)
}

// NewPutRequest creates put request
func (co *ClientConn) NewPutRequest(path string, contentFormat MediaType, body io.Reader) (Message, error) {
	return co.commander.NewPutRequest(path, contentFormat, body)
}

// NewDeleteRequest creates delete request
func (co *ClientConn) NewDeleteRequest(path string) (Message, error) {
	return co.commander.NewDeleteRequest(path)
}

func (co *ClientConn) WriteMsg(m Message) error {
	return co.commander.WriteMsgWithContext(context.Background(), m)
}

// WriteContextMsg sends direct a message through the connection
func (co *ClientConn) WriteMsgWithContext(ctx context.Context, m Message) error {
	return co.commander.WriteMsgWithContext(ctx, m)
}

// Ping send a ping message and wait for a pong response
func (co *ClientConn) Ping(timeout time.Duration) error {
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	return co.PingWithContext(ctx)
}

// Ping send a ping message and wait for a pong response
func (co *ClientConn) PingWithContext(ctx context.Context) error {
	return co.commander.PingWithContext(ctx)
}

// GetContext retrieve the resource identified by the request path
func (co *ClientConn) Get(path string) (Message, error) {
	return co.GetWithContext(context.Background(), path)
}

func (co *ClientConn) GetWithContext(ctx context.Context, path string) (Message, error) {
	if co.multicast {
		return nil, ErrNotSupported
	}
	return co.commander.GetWithContext(ctx, path)
}

func (co *ClientConn) Post(path string, contentFormat MediaType, body io.Reader) (Message, error) {
	return co.PostWithContext(context.Background(), path, contentFormat, body)
}

// Post update the resource identified by the request path
func (co *ClientConn) PostWithContext(ctx context.Context, path string, contentFormat MediaType, body io.Reader) (Message, error) {
	if co.multicast {
		return nil, ErrNotSupported
	}
	return co.commander.PostWithContext(ctx, path, contentFormat, body)
}

func (co *ClientConn) Put(path string, contentFormat MediaType, body io.Reader) (Message, error) {
	return co.PutWithContext(context.Background(), path, contentFormat, body)
}

// PutContext create the resource identified by the request path
func (co *ClientConn) PutWithContext(ctx context.Context, path string, contentFormat MediaType, body io.Reader) (Message, error) {
	if co.multicast {
		return nil, ErrNotSupported
	}
	return co.commander.PutWithContext(ctx, path, contentFormat, body)
}

func (co *ClientConn) Delete(path string) (Message, error) {
	return co.DeleteWithContext(context.Background(), path)
}

// Delete delete the resource identified by the request path
func (co *ClientConn) DeleteWithContext(ctx context.Context, path string) (Message, error) {
	if co.multicast {
		return nil, ErrNotSupported
	}
	return co.commander.DeleteWithContext(ctx, path)
}

func (co *ClientConn) Observe(path string, observeFunc func(req *Request)) (*Observation, error) {
	return co.ObserveWithContext(context.Background(), path, observeFunc)
}

func (co *ClientConn) ObserveWithContext(ctx context.Context, path string, observeFunc func(req *Request)) (*Observation, error) {
	if co.multicast {
		return nil, ErrNotSupported
	}
	return co.commander.ObserveWithContext(ctx, path, observeFunc)
}

// Close close connection
func (co *ClientConn) Close() error {
	var err error
	if co.srv != nil {
		err = co.srv.Shutdown()
	} else {
		err = co.commander.Close()
	}
	if err != nil {
		return err
	}
	if co.shutdownSync != nil {
		err = <-co.shutdownSync
	}
	return err
}

// Sequence discontinuously unique growing number for connection.
func (co *ClientConn) Sequence() uint64 {
	return co.commander.Sequence()
}

// Dial connects to the address on the named network.
func Dial(network, address string) (*ClientConn, error) {
	client := Client{Net: network}
	return client.DialWithContext(context.Background(), address)
}

// DialTimeout acts like Dial but takes a timeout.
func DialTimeout(network, address string, timeout time.Duration) (*ClientConn, error) {
	client := Client{Net: network, DialTimeout: timeout}
	return client.DialWithContext(context.Background(), address)
}

func fixNetTLS(network string) string {
	if !strings.HasSuffix(network, "-tls") {
		network += "-tls"
	}
	return network
}

// DialWithTLS connects to the address on the named network with TLS.
func DialWithTLS(network, address string, tlsConfig *tls.Config) (conn *ClientConn, err error) {
	client := Client{Net: fixNetTLS(network), TLSConfig: tlsConfig}
	return client.DialWithContext(context.Background(), address)
}

// DialTimeoutWithTLS acts like DialWithTLS but takes a timeout.
func DialTimeoutWithTLS(network, address string, tlsConfig *tls.Config, timeout time.Duration) (conn *ClientConn, err error) {
	client := Client{Net: fixNetTLS(network), DialTimeout: timeout, TLSConfig: tlsConfig}
	return client.DialWithContext(context.Background(), address)
}