summaryrefslogtreecommitdiff
path: root/src/globster/hub/hub.go
blob: f17c2e73b07867693342762f7a73be05f1220b1f (plain)
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
package hub

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

// TODO: Do write()s in a separate goroutine.

type hub struct {
	name string
	ses  *tanja.Session
	// Only set in the connected state, nil otherwise.
	conn net.Conn
	// 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{}
}

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 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) {
	s.ses.Sendt(false, append(tanja.Tup("hub", s.name, "Disconnected"), r...))

	if s.connCh != nil {
		// 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 {
		// If we're connected, close the connection.
		go func(c net.Conn) {
			c.Close()
			m.Close()
		}(s.conn)
		s.conn = nil

	} else {
		m.Close()
	}
}

func (s *hub) connected(r interface{}) {
	s.connCh = nil
	switch v := r.(type) {
	case net.Conn:
		s.conn = v
		s.ses.Send(false, "hub", s.name, "Connected", s.conn.RemoteAddr().String())
		adc.NewWriter(s.conn).Write(&adc.Message{Header: adc.SUP.SetType('I')})
	case error:
		s.ses.Send(false, "hub", s.name, "Disconnected", v.Error())
	}
}

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")
}