libnmdc/HubConnection.go

package libnmdc

import (
	"crypto/tls"
	"fmt"
	"net"
	"strconv"
	"strings"
	"sync"
	"time"
)

type HubConnection struct {
	// Supplied parameters
	Hco *HubConnectionOptions

	// Current remote status
	HubName  string
	State    ConnectionState
	users    map[string]UserInfo
	userLock sync.RWMutex

	// Streamed events
	processEvent func(HubEvent)
	OnEvent      chan HubEvent

	// Private state
	conn             net.Conn // this is an interface
	connValid        bool
	sentOurHello     bool
	autoReconnect    bool
	lastDataRecieved time.Time

	supports map[string]struct{}
}

// Thread-safe user accessor.
func (this *HubConnection) Users(cb func(*map[string]UserInfo) error) error {
	this.userLock.Lock()
	defer this.userLock.Unlock()

	return cb(&this.users)
}

func (this *HubConnection) SayPublic(message string) {
	this.SayRaw("<" + this.Hco.Self.Nick + "> " + Escape(message) + "|")
}

func (this *HubConnection) SayPrivate(recipient string, message string) {
	this.SayRaw("$To: " + recipient + " From: " + this.Hco.Self.Nick + " $<" + this.Hco.Self.Nick + "> " + Escape(message) + "|")
}

func (this *HubConnection) SayInfo() {
	this.SayRaw(this.Hco.Self.toMyINFO() + "|")
}

func (this *HubConnection) UserExists(nick string) bool {
	this.userLock.RLock()
	defer this.userLock.RUnlock()

	_, already_existed := this.users[nick]
	return already_existed
}

func (this *HubConnection) UserCount() int {
	this.userLock.RLock()
	defer this.userLock.RUnlock()

	return len(this.users)
}

func (this *HubConnection) userJoined_NameOnly(nick string) {
	if !this.UserExists(nick) {

		this.userLock.Lock()
		this.users[nick] = *NewUserInfo(nick)
		this.userLock.Unlock() // Don't lock over a processEvent boundary

		this.processEvent(HubEvent{EventType: EVENT_USER_JOINED, Nick: nick})
	}
}

func (this *HubConnection) userJoined_Full(uinf *UserInfo) {
	// n.b. also called when we get a replacement MyINFO for someone
	this.userLock.Lock()
	_, userExisted := this.users[uinf.Nick] // don't use UserExists as it would deadlock the mutex
	this.users[uinf.Nick] = *uinf
	this.userLock.Unlock() // Don't lock over a processEvent boundary

	if !userExisted {
		this.processEvent(HubEvent{EventType: EVENT_USER_JOINED, Nick: uinf.Nick})
	} else {
		this.processEvent(HubEvent{EventType: EVENT_USER_UPDATED_INFO, Nick: uinf.Nick})
	}
}

// SayRaw sends raw bytes over the TCP socket. Callers should add the protocol
// terminating character `|` themselves.
// Note that protocol messages are transmitted on the caller thread, not from
// any internal libnmdc thread.
func (this *HubConnection) SayRaw(protocolCommand string) error {
	if this.connValid {
		_, err := this.conn.Write([]byte(protocolCommand))
		return err
	} else {
		return ErrNotConnected
	}
}

func (this *HubConnection) processProtocolMessage(message string) {

	// Zero-length protocol message
	// ````````````````````````````
	if len(message) == 0 {
		return
	}

	// Public chat
	// ```````````
	if rx_publicChat.MatchString(message) {
		pubchat_parts := rx_publicChat.FindStringSubmatch(message)
		this.processEvent(HubEvent{EventType: EVENT_PUBLIC, Nick: pubchat_parts[1], Message: Unescape(pubchat_parts[2])})
		return
	}

	// System messages
	// ```````````````
	if message[0] != '$' {
		this.processEvent(HubEvent{EventType: EVENT_SYSTEM_MESSAGE_FROM_HUB, Nick: this.HubName, Message: Unescape(message)})
		return
	}

	// Protocol messages
	// `````````````````

	commandParts := strings.SplitN(message, " ", 2)
	switch commandParts[0] {

	case "$Lock":
		this.SayRaw("$Supports NoHello NoGetINFO UserCommand UserIP2 QuickList ChatOnly|" +
			"$Key " + unlock([]byte(commandParts[1])) + "|")
		this.sentOurHello = false

	case "$Hello":
		if commandParts[1] == this.Hco.Self.Nick && !this.sentOurHello {
			this.SayRaw("$Version 1,0091|")
			this.SayRaw("$GetNickList|")
			this.SayInfo()
			this.sentOurHello = true

		} else {
			this.userJoined_NameOnly(commandParts[1])

		}

	case "$HubName":
		this.HubName = commandParts[1]
		this.processEvent(HubEvent{EventType: EVENT_HUBNAME_CHANGED, Nick: commandParts[1]})

	case "$ValidateDenide": // sic
		if len(this.Hco.NickPassword) > 0 {
			this.processEvent(HubEvent{EventType: EVENT_SYSTEM_MESSAGE_FROM_CONN, Message: "Incorrect password."})
		} else {
			this.processEvent(HubEvent{EventType: EVENT_SYSTEM_MESSAGE_FROM_CONN, Message: "Nick already in use."})
		}

	case "$HubIsFull":
		this.processEvent(HubEvent{EventType: EVENT_SYSTEM_MESSAGE_FROM_CONN, Message: "Hub is full."})

	case "$BadPass":
		this.processEvent(HubEvent{EventType: EVENT_SYSTEM_MESSAGE_FROM_CONN, Message: "Incorrect password."})

	case "$GetPass":
		if len(this.Hco.NickPassword) == 0 {
			// We've got a problem. MyPass with no arguments is a syntax error with no message = instant close
			// Just drop the connection
			this.processEvent(HubEvent{EventType: EVENT_SYSTEM_MESSAGE_FROM_CONN, Message: "This account is passworded."})
			this.Disconnect()
		} else {
			this.SayRaw("$MyPass " + Escape(this.Hco.NickPassword) + "|")
		}

	case "$Quit":
		this.userLock.Lock()
		delete(this.users, commandParts[1])
		this.userLock.Unlock() // Don't lock over a processEvent boundary

		this.processEvent(HubEvent{EventType: EVENT_USER_PART, Nick: commandParts[1]})

	case "$MyINFO":
		u := UserInfo{}
		err := u.fromMyINFO(commandParts[1])
		if err != nil {
			this.processEvent(HubEvent{EventType: EVENT_DEBUG_MESSAGE, Message: err.Error()})
			return
		}

		this.userJoined_Full(&u)

	case "$NickList":
		nicklist := strings.Split(commandParts[1], "$$")
		for _, nick := range nicklist {
			if len(nick) > 0 {
				this.userJoined_NameOnly(nick)
			}
		}

	case "$OpList":
		oplist := strings.Split(commandParts[1], "$$")
		opmap := map[string]struct{}{}

		// Organise/sort the list, and ensure we're not meeting an operator for
		//  the first time
		for _, nick := range oplist {
			if len(nick) > 0 {
				opmap[nick] = struct{}{}
				this.userJoined_NameOnly(nick) // assert existence; noop otherwise
			}
		}

		// Mark all mentioned nicks as being operators, and all unmentioned nicks
		//  as being /not/ an operator. (second pass minimises RW mutex use)
		func() {
			this.userLock.Lock()
			defer this.userLock.Unlock()

			for nick, userinfo := range this.users {
				_, isop := opmap[nick]

				userinfo.IsOperator = isop
				this.users[nick] = userinfo
			}
		}()

	case "$To:":
		valid := false
		if rx_incomingTo.MatchString(commandParts[1]) {
			txparts := rx_incomingTo.FindStringSubmatch(commandParts[1])
			if txparts[1] == this.Hco.Self.Nick && txparts[2] == txparts[3] {
				this.processEvent(HubEvent{EventType: EVENT_PRIVATE, Nick: txparts[2], Message: Unescape(txparts[4])})
				valid = true
			}
		}

		if !valid {
			this.processEvent(HubEvent{EventType: EVENT_DEBUG_MESSAGE, Message: "Malformed private message '" + commandParts[1] + "'"})
		}

	case "$UserIP":
		this.userLock.Lock()

		pairs := strings.Split(commandParts[1], "$$")
		notifyOfUpdate := make([]string, 0, len(pairs))

	nextIPPair:
		for _, pair := range pairs {
			parts := strings.SplitN(pair, " ", 2)
			if len(parts) != 2 {
				// ????
				continue nextIPPair
			}

			ip2nick := parts[0]
			ip2addr := parts[1]

			uinfo, ok := this.users[ip2nick]
			if !ok {
				this.processEvent(HubEvent{EventType: EVENT_DEBUG_MESSAGE, Message: "Recieved IP '" + ip2addr + "' for unknown user '" + ip2nick + "'"})
				continue nextIPPair
			}

			if uinfo.IPAddress != ip2addr {
				uinfo.IPAddress = ip2addr
				notifyOfUpdate = append(notifyOfUpdate, ip2nick)
				this.users[ip2nick] = uinfo
			}
		}

		this.userLock.Unlock()

		for _, nick := range notifyOfUpdate {
			this.processEvent(HubEvent{EventType: EVENT_USER_UPDATED_INFO, Nick: nick})
		}

	case "$ForceMove":
		this.Hco.Address = HubAddress(commandParts[1])
		this.conn.Close() // we'll reconnect onto the new address

	case "$UserCommand":
		// $UserCommand 1 1 Group chat\New group chat$<%[mynick]> !groupchat_new&#124;|
		if rx_userCommand.MatchString(commandParts[1]) {
			usc := rx_userCommand.FindStringSubmatch(commandParts[1])

			typeInt, _ := strconv.Atoi(usc[1])
			contextInt, _ := strconv.Atoi(usc[2])

			uscStruct := UserCommand{
				Type:    UserCommandType(typeInt),
				Context: UserCommandContext(contextInt),
				Message: usc[3],
				Command: Unescape(usc[4]),
			}

			this.processEvent(HubEvent{EventType: EVENT_USERCOMMAND, UserCommand: &uscStruct})

		} else {
			this.processEvent(HubEvent{EventType: EVENT_DEBUG_MESSAGE, Message: "Malformed usercommand '" + commandParts[1] + "'"})
		}

	case "$Supports":
		this.supports = make(map[string]struct{})
		for _, s := range strings.Split(commandParts[1], " ") {
			this.supports[s] = struct{}{}
		}

		if !this.sentOurHello {

			// Need to log in.
			// If the hub supports QuickList, we can skip one network roundtrip
			if _, ok := this.supports["QuickList"]; ok {
				this.SayInfo()
				this.SayRaw("$GetNickList|")
			} else {
				this.SayRaw("$ValidateNick " + Escape(this.Hco.Self.Nick) + "|")
			}

			// This also counts as the end of the handshake from our POV. Consider
			// ourselves logged in
			this.sentOurHello = true
			if this.State != CONNECTIONSTATE_CONNECTED {
				this.processEvent(HubEvent{EventType: EVENT_CONNECTION_STATE_CHANGED, StateChange: CONNECTIONSTATE_CONNECTED})
				this.State = CONNECTIONSTATE_CONNECTED
			}

		}

	// IGNORABLE COMMANDS
	case "$HubTopic":
	case "$Search":
	case "$ConnectToMe":

	default:
		this.processEvent(HubEvent{EventType: EVENT_DEBUG_MESSAGE, Message: "Unhandled protocol command '" + commandParts[0] + "'"})

	}
}

func (this *HubConnection) Disconnect() {
	this.autoReconnect = false
	if this.conn != nil {
		this.conn.Close()
	}
	// A CONNECTIONSTATE_DISCONNECTED message will be emitted by the worker.
}

func (this *HubConnection) worker() {
	var fullBuffer string
	var err error = nil
	var nbytes int = 0

	for {

		// If we're not connected, attempt reconnect
		if this.conn == nil {

			fullBuffer = "" // clear

			if this.Hco.Address.IsSecure() {
				this.conn, err = tls.Dial("tcp", this.Hco.Address.GetHostOnly(), &tls.Config{
					InsecureSkipVerify: this.Hco.SkipVerifyTLS,
				})
			} else {
				this.conn, err = net.Dial("tcp", this.Hco.Address.GetHostOnly())
			}

			if err != nil {
				this.State = CONNECTIONSTATE_DISCONNECTED
				this.connValid = false

			} else {
				this.State = CONNECTIONSTATE_CONNECTING
				this.connValid = true
				this.processEvent(HubEvent{EventType: EVENT_CONNECTION_STATE_CHANGED, StateChange: CONNECTIONSTATE_CONNECTING})

			}
		}

		// Read from socket into our local buffer (blocking)
		if this.connValid {
			readBuff := make([]byte, 1024)
			this.conn.SetReadDeadline(time.Now().Add(SEND_KEEPALIVE_EVERY))
			nbytes, err = this.conn.Read(readBuff)

			if checkIsNetTimeout(err) {
				// No data before read deadline
				err = nil

				// Send KA packet
				_, err = this.conn.Write([]byte("|"))
			}

			if nbytes > 0 {
				this.lastDataRecieved = time.Now()
				fullBuffer += string(readBuff[0:nbytes])
			}
		}

		// Attempt to parse a message block
		for len(fullBuffer) > 0 {
			for len(fullBuffer) > 0 && fullBuffer[0] == '|' {
				fullBuffer = fullBuffer[1:]
			}
			protocolMessage := rx_protocolMessage.FindString(fullBuffer)
			if len(protocolMessage) > 0 {
				this.processProtocolMessage(protocolMessage[:len(protocolMessage)-1])
				fullBuffer = fullBuffer[len(protocolMessage):]
			} else {
				break
			}
		}

		if err == nil && time.Now().Sub(this.lastDataRecieved) > RECONNECT_IF_NO_DATA_RECIEVED_IN {
			err = fmt.Errorf("No packets recieved since %s, connection presumed lost", this.lastDataRecieved.Format(time.RFC3339))
		}

		// Maybe we disconnected
		// Perform this check *last*, to ensure we've had a final shot at
		//  clearing out any queued messages
		if err != nil {
			this.State = CONNECTIONSTATE_DISCONNECTED
			this.conn = nil
			this.connValid = false
			this.processEvent(HubEvent{EventType: EVENT_CONNECTION_STATE_CHANGED, StateChange: CONNECTIONSTATE_DISCONNECTED, Message: err.Error()})

			if this.autoReconnect {
				time.Sleep(AUTO_RECONNECT_AFTER) // Wait before reconnect
				continue
			} else {
				return // leave the worker for good
			}
		}

	}

}