path: root/src/globster/hub/hub.go

package hub

import (
	"blicky.net/asyncwr"
	"blicky.net/tanja"
	"errors"
	"globster/adc"
	"net"
	"time"
)

type hub struct {
	name string
	ses  *tanja.Session
	// Only set in the connected state, nil otherwise.
	conn net.Conn
	wr   *adc.Writer
	rd   <-chan *adc.Message
	// Channel for the asynchronous connect. Only set in the connecting state,
	// otherwise nil. Only one value is written to the channel, a net.Conn on
	// success or an error otherwise.
	connCh <-chan interface{}
	// ADC info (only useful if conn != nil)
	state adc.State
	mysid adc.FourCC
}

func newHub(node *tanja.Node, name string) *hub {
	s := &hub{name: name, ses: node.Session()}

	s.ses.RegRPC(s, func(s string) tanja.Tuple {
		if s[0] != 'T' {
			return nil
		}
		return tanja.Tup("hub", name, s[1:])
	})

	s.ses.Send(false, "hub", name, "Created")

	go func() {
		for {
			select {
			case m, ok := <-s.ses.Chan():
				if !ok {
					return
				}
				m.Dispatch()
			case m := <-s.rd:
				if m == nil {
					s.rd = nil
				} else if err := adc.HandleHub(m, s.state, s); err != nil {
					s.ses.Send(false, "hub", s.name, "disconnect", err.Error())
				}
			case c := <-s.connCh:
				s.connected(c)
			}
		}
		s.ses.Run()
	}()
	return s
}

func (s *hub) Texists(m *tanja.Message) {
	m.Reply(s.name)
	m.Close()
}

func (s *hub) Tconnect(addr string) {
	if s.conn != nil || s.connCh != nil {
		return
	}
	ch := make(chan interface{}, 2)
	s.connCh = ch
	go func() {
		c, e := net.DialTimeout("tcp", addr, 30*time.Second)
		if e != nil {
			ch <- e
		} else {
			ch <- c
		}
	}()
}

// Sending 'disconnect' will immediately put the hub into disconnected state,
// but the actual disconnect from the hub may take a while. The "Disconnected"
// notification is sent immediately, but the return path is only closed after
// the disconnect has been completed.
func (s *hub) Tdisconnect(m *tanja.Message, r ...tanja.Element) {
	broadcast := false

	if s.connCh != nil {
		broadcast = true
		// If we're in the connecting state, set connCh to nil so we won't be
		// getting its result. In case the connection was successful, make sure
		// to immediately disconnect again.
		go func(ch <-chan interface{}) {
			if c, ok := (<-ch).(net.Conn); ok {
				c.Close()
			}
		}(s.connCh)
		s.connCh = nil
		m.Close()

	} else if s.conn != nil {
		broadcast = true
		// If we're connected, close the connection.
		go func(c net.Conn) {
			c.Close()
			m.Close()
		}(s.conn)
		s.conn = nil
		s.wr = nil
		// Also drain the reader channel (to make sure the reader goroutine doesn't deadlock)
		if s.rd != nil {
			go func(ch <-chan *adc.Message) {
				for _ = range ch {
				}
			}(s.rd)
			s.rd = nil
		}
		s.state = adc.InvalidState

	} else {
		m.Close()
	}

	if broadcast {
		s.ses.Sendt(false, append(tanja.Tup("hub", s.name, "Disconnected"), r...))
	}
}

func (s *hub) Tchat(user tanja.Element, me bool, msg string) {
	if s.state != adc.NORMAL {
		return
	}
	var cmd *adc.Message
	if user.WC() {
		cmd = adc.NewMessage('B', adc.MSG)
	} else if adc.Arg(user.String()).IsSID() {
		// TODO: type = E?
		cmd = adc.NewMessage('D', adc.MSG).SetDest(adc.Arg(user.String()).FourCC())
	} else {
		return
	}
	cmd.SetSrc(s.mysid).AddPos(msg)
	if me {
		cmd.AddNamed("ME", "1")
	}
	s.wr.Write(cmd)
}

func (s *hub) connected(r interface{}) {
	s.connCh = nil
	if e, ok := r.(error); ok {
		s.ses.Send(false, "hub", s.name, "Disconnected", e.Error())
		return
	}

	// Successful connect
	s.conn = r.(net.Conn)
	s.ses.Send(false, "hub", s.name, "Connected", s.conn.RemoteAddr().String())
	s.wr = adc.NewWriter(asyncwr.New(s.conn, 1024*1024))

	// Start reader goroutine
	ch := make(chan *adc.Message, 10)
	s.rd = ch
	go func(rd *adc.Reader) {
		for {
			m, err := rd.Read()
			if err != nil {
				s.ses.Send(false, "hub", s.name, "disconnect", err.Error())
				break
			} else if m != nil {
				ch <- m
			}
		}
		close(ch)
	}(adc.NewReader(s.conn, 100*1024))

	// Initiate handshake
	s.state = adc.PROTOCOL
	s.wr.Write(adc.NewMessage('H', adc.SUP).AddNamed("AD", "BASE").AddNamed("AD", "TIGR"))
}

func (s *hub) AdcSUP(ad []string, rm []string) error {
	if s.state != adc.PROTOCOL || len(rm) > 0 {
		return errors.New("Dynamic changing of protocol features is not supported")
	}
	hasbase, hastigr := false, false
	for _, v := range ad {
		if string(v) == "BASE" {
			hasbase = true
		} else if string(v) == "TIGR" {
			hastigr = true
		}
	}
	if !hasbase || !hastigr {
		return errors.New("Hub does not support BASE or TIGR")
	}
	return nil
}

func (s *hub) AdcSID(sid adc.FourCC) error {
	s.mysid = sid
	s.state = adc.IDENTIFY

	// Send our initial INF
	res := s.ses.Send(true, "userlist", s.name, "get", "",
		"PID", "CID", "Name", "Client", "IP4", "Slots", "ShareSize", "ShareFiles",
		"HubsNorm", "HubsReg", "HubsOp", "Desc", "EMail")
	nfo := <-res.Chan()
	res.Close()
	if nfo == nil {
		return errors.New("No local user info")
	}
	n := nfo[0].Map()
	s.wr.Write(adc.NewMessage('B', adc.INF).SetSrc(s.mysid).
		AddNamed("NI", n["Name"].String()).
		AddNamed("ID", n["CID"].String()).
		AddNamed("PD", n["PID"].String()).
		AddNamed("I4", n["IP4"].String()).
		AddNamed("U4", "").
		AddNamed("SU", "").
		AddNamed("SL", n["Slots"].String()).
		AddNamed("SF", n["ShareFiles"].String()).
		AddNamed("SS", n["ShareSize"].String()).
		AddNamed("HN", n["HubsNorm"].String()).
		AddNamed("HR", n["HubsReg"].String()).
		AddNamed("HO", n["HubsOp"].String()).
		AddNamed("EM", n["EMail"].String()).
		AddNamed("DE", n["Desc"].String()).
		AddNamed("VE", n["Client"].String()))

	// TODO: We're not actually in the NORMAL yet, but this works for debugging purposes.
	s.state = adc.NORMAL
	return nil
}

func (s *hub) close() {
	r := s.ses.Send(true, "hub", s.name, "disconnect")
	<-r.Chan()
	r.Close()

	s.ses.Close()
	s.ses.Send(false, "hub", s.name, "Closed")
}