micro/tunnel/session.go

291 lines
5.2 KiB
Go
Raw Normal View History

2019-08-06 13:45:25 +03:00
package tunnel
2019-08-07 20:44:33 +03:00
import (
"errors"
2019-08-30 22:05:00 +03:00
"io"
"time"
2019-08-06 13:45:25 +03:00
2019-08-07 20:44:33 +03:00
"github.com/micro/go-micro/transport"
2019-08-11 20:11:33 +03:00
"github.com/micro/go-micro/util/log"
2019-08-07 20:44:33 +03:00
)
2019-08-06 13:45:25 +03:00
2019-08-30 22:05:00 +03:00
// session is our pseudo session for transport.Socket
type session struct {
2019-09-03 17:56:37 +03:00
// the tunnel id
tunnel string
2019-08-30 22:05:00 +03:00
// the channel name
channel string
2019-08-07 20:44:33 +03:00
// 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
// if the discovery worked
discovered bool
// if the session was accepted
accepted bool
2019-08-30 22:05:00 +03:00
// outbound marks the session as outbound dialled connection
2019-08-14 19:14:39 +03:00
outbound bool
2019-08-30 22:05:00 +03:00
// lookback marks the session as a loopback on the inbound
loopback bool
// if the session is multicast
multicast bool
// if the session is broadcast
broadcast bool
// the timeout
timeout time.Duration
2019-08-29 14:42:27 +03:00
// the link on which this message was received
link string
2019-08-30 22:05:00 +03:00
// the error response
errChan chan error
2019-08-06 13:45:25 +03:00
}
2019-08-07 20:44:33 +03:00
// message is sent over the send channel
type message struct {
2019-08-29 14:42:27 +03:00
// type of message
typ string
2019-08-07 20:44:33 +03:00
// tunnel id
2019-09-03 17:56:37 +03:00
tunnel string
2019-08-30 22:05:00 +03:00
// channel name
channel string
2019-08-07 20:44:33 +03:00
// the session id
session string
2019-08-14 19:14:39 +03:00
// outbound marks the message as outbound
outbound bool
// loopback marks the message intended for loopback
loopback bool
// whether to send as multicast
multicast bool
// broadcast sets the broadcast type
broadcast bool
2019-08-29 14:42:27 +03:00
// the link to send the message on
link string
2019-08-07 20:44:33 +03:00
// transport data
data *transport.Message
2019-08-30 22:05:00 +03:00
// the error channel
errChan chan error
2019-08-07 20:44:33 +03:00
}
2019-08-30 22:05:00 +03:00
func (s *session) Remote() string {
2019-08-07 20:44:33 +03:00
return s.remote
}
2019-08-30 22:05:00 +03:00
func (s *session) Local() string {
2019-08-07 20:44:33 +03:00
return s.local
2019-08-06 13:45:25 +03:00
}
2019-09-11 22:07:43 +03:00
func (s *session) Link() string {
return s.link
}
2019-08-30 22:05:00 +03:00
func (s *session) Id() string {
return s.session
2019-08-07 20:44:33 +03:00
}
2019-08-30 22:05:00 +03:00
func (s *session) Channel() string {
return s.channel
2019-08-07 20:44:33 +03:00
}
2019-09-04 14:18:37 +03:00
// newMessage creates a new message based on the session
2019-09-04 14:16:31 +03:00
func (s *session) newMessage(typ string) *message {
return &message{
typ: typ,
tunnel: s.tunnel,
channel: s.channel,
session: s.session,
outbound: s.outbound,
loopback: s.loopback,
multicast: s.multicast,
link: s.link,
errChan: s.errChan,
}
2019-09-04 14:16:31 +03:00
}
// Open will fire the open message for the session. This is called by the dialler.
func (s *session) Open() error {
// create a new message
msg := s.newMessage("open")
// send open message
s.send <- msg
// wait for an error response for send
select {
case err := <-msg.errChan:
if err != nil {
return err
}
case <-s.closed:
return io.EOF
}
// we don't wait on multicast
if s.multicast {
s.accepted = true
return nil
}
// now wait for the accept
select {
case msg = <-s.recv:
if msg.typ != "accept" {
log.Debugf("Received non accept message in Open %s", msg.typ)
return errors.New("failed to connect")
}
// set to accepted
s.accepted = true
// set link
s.link = msg.link
case <-time.After(s.timeout):
return ErrDialTimeout
case <-s.closed:
return io.EOF
}
return nil
}
2019-09-04 13:58:03 +03:00
// Accept sends the accept response to an open message from a dialled connection
func (s *session) Accept() error {
2019-09-04 14:16:31 +03:00
msg := s.newMessage("accept")
// send the accept message
select {
case <-s.closed:
return io.EOF
case s.send <- msg:
return nil
}
// wait for send response
select {
case err := <-s.errChan:
if err != nil {
return err
}
case <-s.closed:
return io.EOF
}
return nil
}
2019-09-04 13:58:03 +03:00
// Announce sends an announcement to notify that this session exists. This is primarily used by the listener.
func (s *session) Announce() error {
2019-09-04 14:16:31 +03:00
msg := s.newMessage("announce")
// we don't need an error back
msg.errChan = nil
// announce to all
msg.broadcast = true
2019-09-04 14:16:31 +03:00
// we don't need the link
msg.link = ""
2019-09-04 13:58:03 +03:00
select {
case s.send <- msg:
return nil
case <-s.closed:
return io.EOF
}
}
// Send is used to send a message
2019-08-30 22:05:00 +03:00
func (s *session) Send(m *transport.Message) error {
2019-08-07 20:44:33 +03:00
select {
case <-s.closed:
return io.EOF
2019-08-07 20:44:33 +03:00
default:
// no op
}
2019-08-11 20:11:33 +03: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-09-04 14:16:31 +03:00
// create a new message
msg := s.newMessage("session")
// set the data
msg.data = data
2019-09-04 14:16:31 +03:00
// if multicast don't set the link
if s.multicast {
msg.link = ""
}
2019-08-11 20:11:33 +03:00
log.Debugf("Appending %+v to send backlog", msg)
2019-09-04 13:58:03 +03:00
// send the actual message
2019-08-11 20:11:33 +03:00
s.send <- msg
2019-08-30 22:05:00 +03:00
// wait for an error response
select {
case err := <-msg.errChan:
return err
case <-s.closed:
return io.EOF
}
2019-08-06 13:45:25 +03:00
return nil
}
2019-09-04 13:58:03 +03:00
// Recv is used to receive a message
2019-08-30 22:05:00 +03:00
func (s *session) Recv(m *transport.Message) error {
2019-08-07 20:44:33 +03:00
select {
case <-s.closed:
2019-08-30 22:05:00 +03:00
return errors.New("session is closed")
2019-08-07 20:44:33 +03:00
default:
// no op
}
// recv from backlog
msg := <-s.recv
2019-08-30 22:05:00 +03:00
// check the error if one exists
select {
case err := <-msg.errChan:
return err
default:
}
2019-08-11 20:11:33 +03:00
log.Debugf("Received %+v from recv backlog", msg)
2019-08-07 20:44:33 +03:00
// set message
*m = *msg.data
// return nil
return nil
2019-08-06 13:45:25 +03:00
}
2019-09-04 13:58:03 +03:00
// Close closes the session by sending a close message
2019-08-30 22:05:00 +03:00
func (s *session) Close() error {
2019-08-07 20:44:33 +03:00
select {
case <-s.closed:
// no op
default:
close(s.closed)
// append to backlog
2019-09-04 14:16:31 +03:00
msg := s.newMessage("close")
// no error response on close
msg.errChan = nil
// send the close message
select {
case s.send <- msg:
default:
}
2019-08-07 20:44:33 +03:00
}
2019-08-07 20:44:33 +03:00
return nil
2019-08-06 13:45:25 +03:00
}