micro/tunnel/session.go

package tunnel

import (
	"errors"
	"io"
	"time"

	"github.com/micro/go-micro/transport"
	"github.com/micro/go-micro/util/log"
)

// session is our pseudo session for transport.Socket
type session struct {
	// the tunnel id
	tunnel string
	// the channel name
	channel 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
	// if the discovery worked
	discovered bool
	// if the session was accepted
	accepted bool
	// outbound marks the session as outbound dialled connection
	outbound bool
	// lookback marks the session as a loopback on the inbound
	loopback bool
	// mode of the connection
	mode Mode
	// the timeout
	timeout time.Duration
	// the link on which this message was received
	link string
	// the error response
	errChan chan error
}

// message is sent over the send channel
type message struct {
	// type of message
	typ string
	// tunnel id
	tunnel string
	// channel name
	channel string
	// the session id
	session string
	// outbound marks the message as outbound
	outbound bool
	// loopback marks the message intended for loopback
	loopback bool
	// mode of the connection
	mode Mode
	// the link to send the message on
	link string
	// transport data
	data *transport.Message
	// the error channel
	errChan chan error
}

func (s *session) Remote() string {
	return s.remote
}

func (s *session) Local() string {
	return s.local
}

func (s *session) Link() string {
	return s.link
}

func (s *session) Id() string {
	return s.session
}

func (s *session) Channel() string {
	return s.channel
}

// newMessage creates a new message based on the session
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,
		mode:     s.mode,
		link:     s.link,
		errChan:  s.errChan,
	}
}

// waitFor waits for the message type required until the timeout specified
func (s *session) waitFor(msgType string, timeout time.Duration) (*message, error) {
	now := time.Now()

	after := func() time.Duration {
		d := time.Since(now)
		// dial timeout minus time since
		wait := timeout - d

		if wait < time.Duration(0) {
			return time.Duration(0)
		}

		return wait
	}

	// wait for the message type
	for {
		select {
		case msg := <-s.recv:
			// ignore what we don't want
			if msg.typ != msgType {
				log.Debugf("Tunnel received non %s message in waiting for %s", msg.typ, msgType)
				continue
			}
			// got the message
			return msg, nil
		case <-time.After(after()):
			return nil, ErrDialTimeout
		case <-s.closed:
			return nil, io.EOF
		}
	}
}

// Discover attempts to discover the link for a specific channel
func (s *session) Discover() error {
	// create a new discovery message for this channel
	msg := s.newMessage("discover")
	msg.mode = Broadcast
	msg.outbound = true
	msg.link = ""

	// send the discovery message
	s.send <- msg

	// set time now
	now := time.Now()

	after := func() time.Duration {
		d := time.Since(now)
		// dial timeout minus time since
		wait := s.timeout - d
		if wait < time.Duration(0) {
			return time.Duration(0)
		}
		return wait
	}

	// wait to hear back about the sent message
	select {
	case <-time.After(after()):
		return ErrDialTimeout
	case err := <-s.errChan:
		if err != nil {
			return err
		}
	}

	var err error

	// set a new dialTimeout
	dialTimeout := after()

	// set a shorter delay for multicast
	if s.mode != Unicast {
		// shorten this
		dialTimeout = time.Millisecond * 500
	}

	// wait for announce
	_, err = s.waitFor("announce", dialTimeout)
	if err != nil {
		return err
	}

	// if its multicast just go ahead because this is best effort
	if s.mode != Unicast {
		s.discovered = true
		s.accepted = true
		return nil
	}

	if err != nil {
		return err
	}

	// set discovered
	s.discovered = true

	return nil
}

// 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
	}

	// don't wait on multicast/broadcast
	if s.mode == Multicast {
		s.accepted = true
		return nil
	}

	// now wait for the accept
	msg, err := s.waitFor("accept", s.timeout)
	if err != nil {
		return err
	}

	// set to accepted
	s.accepted = true
	// set link
	s.link = msg.link

	return nil
}

// Accept sends the accept response to an open message from a dialled connection
func (s *session) Accept() error {
	msg := s.newMessage("accept")

	// send the accept message
	select {
	case <-s.closed:
		return io.EOF
	case s.send <- msg:
		// no op here
	}

	// don't wait on multicast/broadcast
	if s.mode == Multicast {
		return nil
	}

	// wait for send response
	select {
	case err := <-s.errChan:
		if err != nil {
			return err
		}
	case <-s.closed:
		return io.EOF
	}

	return nil
}

// Announce sends an announcement to notify that this session exists. This is primarily used by the listener.
func (s *session) Announce() error {
	msg := s.newMessage("announce")
	// we don't need an error back
	msg.errChan = nil
	// announce to all
	msg.mode = Broadcast
	// we don't need the link
	msg.link = ""

	select {
	case s.send <- msg:
		return nil
	case <-s.closed:
		return io.EOF
	}
}

// Send is used to send a message
func (s *session) Send(m *transport.Message) error {
	select {
	case <-s.closed:
		return io.EOF
	default:
		// no op
	}

	// make copy
	data := &transport.Message{
		Header: make(map[string]string),
		Body:   m.Body,
	}

	for k, v := range m.Header {
		data.Header[k] = v
	}

	// create a new message
	msg := s.newMessage("session")
	// set the data
	msg.data = data

	// if multicast don't set the link
	if s.mode == Multicast {
		msg.link = ""
	}

	log.Debugf("Appending %+v to send backlog", msg)
	// send the actual message
	s.send <- msg

	// wait for an error response
	select {
	case err := <-msg.errChan:
		return err
	case <-s.closed:
		return io.EOF
	}
}

// Recv is used to receive a message
func (s *session) Recv(m *transport.Message) error {
	select {
	case <-s.closed:
		return errors.New("session is closed")
	default:
		// no op
	}
	// recv from backlog
	msg := <-s.recv

	// check the error if one exists
	select {
	case err := <-msg.errChan:
		return err
	default:
	}

	log.Debugf("Received %+v from recv backlog", msg)
	// set message
	*m = *msg.data
	// return nil
	return nil
}

// Close closes the session by sending a close message
func (s *session) Close() error {
	select {
	case <-s.closed:
		// no op
	default:
		close(s.closed)

		// append to backlog
		msg := s.newMessage("close")
		// no error response on close
		msg.errChan = nil

		// send the close message
		select {
		case s.send <- msg:
		default:
		}
	}

	return nil
}