libnmdc/HubConnection.go

439 lines
12 KiB
Go

package libnmdc
import (
"crypto/tls"
"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
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 {
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
}
}
// 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
}
}
}
}