package core import ( "errors" protocol "github.com/s-rah/go-ricochet" "github.com/ricochet-im/ricochet-go/core/utils" "github.com/ricochet-im/ricochet-go/rpc" "log" "math/rand" "sync" "time" ) // XXX should have limits on backlog size/duration type Conversation struct { Contact *Contact mutex sync.Mutex localEntity *ricochet.Entity remoteEntity *ricochet.Entity messages []*ricochet.Message lastSentMessageId uint32 events *utils.Publisher } func NewConversation(contact *Contact, remoteEntity *ricochet.Entity, eventStream *utils.Publisher) *Conversation { return &Conversation{ Contact: contact, localEntity: &ricochet.Entity{IsSelf: true}, remoteEntity: remoteEntity, messages: make([]*ricochet.Message, 0), events: eventStream, } } func (c *Conversation) EventMonitor() utils.Subscribable { return c.events } func (c *Conversation) Messages() []*ricochet.Message { c.mutex.Lock() defer c.mutex.Unlock() re := make([]*ricochet.Message, 0, len(c.messages)) for _, message := range c.messages { re = append(re, message) } return re } func (c *Conversation) Receive(id uint64, timestamp int64, text string) { message := &ricochet.Message{ Sender: c.remoteEntity, Recipient: c.localEntity, Timestamp: timestamp, Identifier: id, Status: ricochet.Message_UNREAD, Text: text, } c.mutex.Lock() defer c.mutex.Unlock() c.messages = append(c.messages, message) event := ricochet.ConversationEvent{ Type: ricochet.ConversationEvent_RECEIVE, Msg: message, } c.events.Publish(event) } func (c *Conversation) UpdateSentStatus(id uint64, success bool) { c.mutex.Lock() defer c.mutex.Unlock() for i := len(c.messages) - 1; i >= 0; i-- { message := c.messages[i] if message.Status != ricochet.Message_SENDING || message.Identifier != id { continue } if success { message.Status = ricochet.Message_DELIVERED } else { message.Status = ricochet.Message_ERROR } event := ricochet.ConversationEvent{ Type: ricochet.ConversationEvent_UPDATE, Msg: message, } c.events.Publish(event) return } log.Printf("Ignoring ack for unknown message id %d", id) } func (c *Conversation) Send(text string) (*ricochet.Message, error) { if len(text) == 0 { return nil, errors.New("Message text is empty") } else if len(text) > 2000 { return nil, errors.New("Message is too long") } c.mutex.Lock() defer c.mutex.Unlock() if c.lastSentMessageId == 0 { // Rand is seeded by Ricochet.Init c.lastSentMessageId = rand.Uint32() } if c.lastSentMessageId++; c.lastSentMessageId == 0 { c.lastSentMessageId++ } message := &ricochet.Message{ Sender: c.localEntity, Recipient: c.remoteEntity, Timestamp: time.Now().Unix(), Identifier: uint64(c.lastSentMessageId), Status: ricochet.Message_QUEUED, Text: text, } // XXX threading mess, and probably deadlockable. Need better API for conn. conn := c.Contact.Connection() if conn != nil { sendMessageToConnection(conn, message) message.Status = ricochet.Message_SENDING } c.messages = append(c.messages, message) event := ricochet.ConversationEvent{ Type: ricochet.ConversationEvent_SEND, Msg: message, } c.events.Publish(event) return message, nil } // Send all messages in the QUEUED state to the contact, if // a connection is available. Should be called after a new // connection is established. func (c *Conversation) SendQueuedMessages() int { c.mutex.Lock() defer c.mutex.Unlock() conn := c.Contact.Connection() if conn == nil { return 0 } sent := 0 for _, message := range c.messages { if message.Status != ricochet.Message_QUEUED { continue } sendMessageToConnection(conn, message) message.Status = ricochet.Message_SENDING sent++ event := ricochet.ConversationEvent{ Type: ricochet.ConversationEvent_UPDATE, Msg: message, } c.events.Publish(event) } return sent } func sendMessageToConnection(conn *protocol.OpenConnection, message *ricochet.Message) { // XXX hardcoded channel IDs, also channel IDs shouldn't be exposed channelId := int32(7) if !conn.Client { channelId++ } // XXX no error handling if conn.GetChannelType(channelId) != "im.ricochet.chat" { conn.OpenChatChannel(channelId) } // XXX no message IDs conn.SendMessage(channelId, message.Text) } // XXX This is inefficient -- it'll usually only be marking the last message // or few messages. Need a better way to know what's unread. func (c *Conversation) MarkReadBeforeMessage(msgId uint64) int { c.mutex.Lock() defer c.mutex.Unlock() marked := 0 for _, message := range c.messages { if message.Status == ricochet.Message_UNREAD { message.Status = ricochet.Message_READ marked++ event := ricochet.ConversationEvent{ Type: ricochet.ConversationEvent_UPDATE, Msg: message, } c.events.Publish(event) } if message.Identifier == msgId && message.Recipient.IsSelf { break } } return marked }