1
0
mirror of https://github.com/go-micro/go-micro.git synced 2024-11-30 08:06:40 +02:00
go-micro/tunnel/socket.go

116 lines
1.9 KiB
Go
Raw Normal View History

2019-08-06 12:45:25 +02:00
package tunnel
2019-08-07 19:44:33 +02:00
import (
"errors"
2019-08-06 12:45:25 +02:00
2019-08-07 19:44:33 +02:00
"github.com/micro/go-micro/transport"
2019-08-11 19:11:33 +02:00
"github.com/micro/go-micro/util/log"
2019-08-07 19:44:33 +02:00
)
2019-08-06 12:45:25 +02:00
2019-08-07 19:44:33 +02:00
// socket is our pseudo socket for transport.Socket
type socket struct {
// socket id based on Micro-Tunnel
id string
// the session id based on Micro.Tunnel-Session
session string
// closed
closed chan bool
// remote addr
remote string
// local addr
local string
// send chan
send chan *message
// recv chan
recv chan *message
// wait until we have a connection
wait chan bool
2019-08-14 18:14:39 +02:00
// outbound marks the socket as outbound
outbound bool
2019-08-06 12:45:25 +02:00
}
2019-08-07 19:44:33 +02:00
// message is sent over the send channel
type message struct {
// tunnel id
id string
// the session id
session string
2019-08-14 18:14:39 +02:00
// outbound marks the message as outbound
outbound bool
2019-08-07 19:44:33 +02:00
// transport data
data *transport.Message
}
func (s *socket) Remote() string {
return s.remote
}
func (s *socket) Local() string {
return s.local
2019-08-06 12:45:25 +02:00
}
2019-08-07 19:44:33 +02:00
func (s *socket) Id() string {
return s.id
}
func (s *socket) Session() string {
return s.session
}
func (s *socket) Send(m *transport.Message) error {
select {
case <-s.closed:
return errors.New("socket is closed")
default:
// no op
}
2019-08-11 19:11:33 +02:00
// make copy
data := &transport.Message{
Header: make(map[string]string),
Body: m.Body,
}
for k, v := range m.Header {
data.Header[k] = v
}
2019-08-07 19:44:33 +02:00
// append to backlog
2019-08-14 18:14:39 +02:00
msg := &message{
id: s.id,
session: s.session,
outbound: s.outbound,
data: data,
}
2019-08-11 19:11:33 +02:00
log.Debugf("Appending %+v to send backlog", msg)
s.send <- msg
2019-08-06 12:45:25 +02:00
return nil
}
2019-08-07 19:44:33 +02:00
func (s *socket) Recv(m *transport.Message) error {
select {
case <-s.closed:
return errors.New("socket is closed")
default:
// no op
}
// recv from backlog
msg := <-s.recv
2019-08-11 19:11:33 +02:00
log.Debugf("Received %+v from recv backlog", msg)
2019-08-07 19:44:33 +02:00
// set message
*m = *msg.data
// return nil
return nil
2019-08-06 12:45:25 +02:00
}
// Close closes the socket
2019-08-07 19:44:33 +02:00
func (s *socket) Close() error {
select {
case <-s.closed:
// no op
default:
close(s.closed)
}
return nil
2019-08-06 12:45:25 +02:00
}