synced/tinode-chat
1
0
Fork 0
mirror of https://github.com/tinode/chat.git synced 2025-03-14 10:05:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

daily commit

Browse Source
This commit is contained in:
Gene Sokolov
2015-12-01 14:57:10 -08:00
parent 2080df197d
commit 72b0c44fda
12 changed files with 205 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
README.md
View File

@ -2,9 +2,10 @@
Instant messaging server. Backend in pure [Go](http://golang.org) ([Affero GPL 3.0](http://www.gnu.org/licenses/agpl-3.0.en.html)), client-side binding in Java for Android and Javascript ([Apache 2.0](http://www.apache.org/licenses/LICENSE-2.0)), persistent storage [RethinkDB](http://rethinkdb.com/), JSON over websocket (long polling is also available). No UI components other than demo apps. Tinode is meant as a replacement for XMPP.
This is alpha-quality software. Bugs should be expected. Version 0.5.
This is alpha-quality software. Bugs should be expected. Version 0.5. Follow [instructions](INSTALL.md) to install.
A demo is (usually) available at [http://api.tinode.co/x/samples/chatdemo.html](http://api.tinode.co/x/samples/chatdemo.html). Login as one of `alice`, `bob`, `carol`, `dave`, `frank`. Password is `<login>123`, e.g. login for `alice` is `alice123`.
A running demo is (usually) available at [http://api.tinode.co/x/samples/chatdemo.html](http://api.tinode.co/x/samples/chatdemo.html). Login as one of `alice`, `bob`, `carol`, `dave`, `frank`. Password is `<login>123`, e.g. login for `alice` is `alice123`.
## Why?
@ -26,7 +27,7 @@ A running demo is (usually) available at [http://api.tinode.co/x/samples/chatdem
* Websocket & long polling transport
* JSON wire protocol
* Server-generated message delivery status
* Support for client-side content caching
* Basic support for client-side message caching
* Blocking users on the server
### Planned
@ -76,6 +77,8 @@ Timestamps are always represented as RFC 3999-formatted string with precision to
Whenever base64 encoding is mentioned, it means base64 URL encoding with padding characters stripped, see RFC 4648.
Server-issued message IDs are base-10 sequential numbers starting at 1. They guaranteed to be unique per topic. Client-assigned message IDs are strings generated by the client. Client should make them unique at least per session. The client-assigned IDs are not interpreted by the server, they are returned to the client as is.
## Connecting to the server
Client establishes a connection to the server over HTTP. Server offers two end points:
@ -118,7 +121,7 @@ To update credentials leave `acc.user` unset.
```js
acc: {
id: "1a2b3", // string, client-provided message id, optional
id: "1a2b3", // string, client-provided message id, optional
user: "new", // string, "new" to create a new user, default: current user, optional
auth: [ // array of authentication schemes to add, update or delete
{
@ -162,8 +165,8 @@ login: {
// scheme, required
expireIn: "24h", // string, login expiration time in Go's time.ParseDuration
// format, see below, optional
tag: "some string" // string, client instance ID; tag is used to support caching,
// optional
ua: "Tinode JS 1.0 (Windows 10)" // string, user agent string identifying client
// software, optional
}
```
Basic authentication scheme expects `secret` to be a string composed of a user name followed by a colon `:` followed by a plan text password.
@ -378,7 +381,7 @@ data: {
// message; could be missing if the message was
// generated by the server
ts: "2015-10-06T18:07:30.038Z", // string, timestamp
seq: 123, // server-issued sequential ID; the lowest possible ID is 1
seq: 123, // integer, server-issued sequential ID
content: { ... } // object, application-defined content exactly as published
// by the user in the {pub} message
}
@ -425,7 +428,7 @@ ctrl: {
"want":"RWP", // string, requested access permission
"given":"RWP" // string, granted access permission
},
lastMsg: "2015-10-29T16:19:15.03Z", // timestamp of the last {data} message
seq: 123, // integer, server-issued id of the last {data} message
public: { ... }, // application-defined data that's available to all topic
// subscribers
private: { ...} // application-deinfed data that's available to the current
@ -433,8 +436,9 @@ ctrl: {
}, // object, topic description, optional
sub: [
{
user: "usr2il9suCbuko", // string, ID of the user this subscription describes
online: "on", // string, current online status of the user with respect to
user: "usr2il9suCbuko", // string, ID of the user this subscription
// describes, absent when querying 'me'
online: "on", // string, current online status of the user with respect to
// the topic, i.e. if the user is listening to messages
updated: "2015-10-24T10:26:09.716Z", // timestamp of the last change in the
// subscription, present only for
@ -444,6 +448,15 @@ ctrl: {
public: { ... }, // application-defined user's 'public' object
private: { ... } // application-defined user's 'private' object, present only
// for the requester's own subscriptions
// The following fields are present only when querying 'me' table
topic: "grp1XUtEhjv6HND", // string, topic this subscription describes
seq: 321, // integer, server-issued id of the last {data} message
with: "usr2il9suCbuko", // string, if this is a P2P topic, peer's ID, optional
seen: { // object, if this is a P2P topic, info on when the peer was last
//online
when: "2015-10-24T10:26:09.716Z", // timestamp
ua: "Tinode/1.0 (Android 5.1)" // string, user agent of peer's client
}
},
...
] // array of objects, topic subscribers, optional
@ -455,7 +468,7 @@ ctrl: {
Tinode uses `{pres}` message to inform users of important events. The following events are tracked by the server and will generate `{pres}` messages provided user has appropriate access permissions:
* A user joins `me`. User receives presence notifications for each of his/her subscriptions: `{pres topic="me" src="<user ID or topic ID>" what="on"}`. Only online status is reported.
* A user joins `me`. User receives presence notifications for each of his/her subscriptions: `{pres topic="me" src="<user ID or topic ID>" what="on", params={ua="..."}}`. Only online status is reported.
* A user came online or went offline. The user triggers this event by joining/leaving the `me` topic. The message is sent to all users who have P2P topics with the first user. Users receive this event on the `me` topic, `src` field contains user ID `src: "usr2il9suCbuko"`, `what` contains `"on"` or `"off"`: `{pres topic="me" src="<user ID>" what="on|off"}`.
* User's `public` is updated. The event is sent to all users who have P2P topics with the first user. Users receive `{pres topic="me" src="<user ID>" what="upd"}`.
* User joins/leaves a topic. This event is sent to other users who currently joined the topic: `{pres topic="<topic name>" src="<user ID>" what="on|off"}`.
@ -467,7 +480,13 @@ Tinode uses `{pres}` message to inform users of important events. The following
pres: {
topic: "grp1XUtEhjv6HND", // string, topic affected by the change, always present
src: "usr2il9suCbuko", // user or topic affected by the change, always present
what: "on" // string, what's changed, always present
what: "on", // string, what's changed, always present
params: { // object, additional context-dependent information, optional
seq: 123, // integer, if "what" is "msg", server-issued ID of the message,
// optional
ua: "Tinode/1.0 (Android 2.2)" // string, if "what" is "on", User-Agent string
// identifying client software, optional
}
}
```
@ -493,7 +512,7 @@ Each user is assigned a unique ID. The IDs are composed as `usr` followed by bas
* public: an application-defined object that describes the user. Anyone who can query user for `public` data.
* private: an application-defined object that is unique to the current user and accessible only by the user.
A user may maintain multiple simultaneous connections (sessions) with the server. Each session can be tagged with a device ID. See [Support for Client-Side Caching][] foir details.
A user may maintain multiple simultaneous connections (sessions) with the server. Each session is tagged with a client-provided User Agent string intended to differentiate client software.
Logging out is not supported by design. If an application needs to change the user, it should open a new connection and authenticate it with the new user credentials.
@ -530,17 +549,13 @@ Topic properties independent of the user making the query:
* defacs: object describing topic's default access mode for authenticated and anonymous users; see [Access control][] for details
* auth: default access mode for authenticated users
* anon: default access for anonymous users
* lastMsg: timestamp when last `{data}` message was sent through the topic
* seq: integer server-issued sequential ID of the latest `{data}` message sent through the topic
* public: an application-defined object that describes the topic. Anyone who can subscribe to topic can receive topic's `public` data.
User-dependent topic properties:
* acs: object describing given user's current access permissions; see [Access control][] for details
* want: access permission requested by this user
* given: access permissions given to this user
* seen: an object describing when the topic was last accessed by the current user from any client instance. This should be useful if the client implements data caching. See [Support for Client-Side Caching][] for more details.
* when": timestamp of the last access
* tag: string provided by the client instance when it accessed the topic.
* seenTag: timestamp when the topic was last accessed from a session with the current client instance. See [Support for Client-Side Caching][] for more details
* private: an application-defined object that is unique to the current user.
Topic usually have subscribers. One the the subscribers may be designated as topic owner (`O` access permission) with full access permissions. The list of subscribers can be queries with a `{get what="sub"}` message. The list of subscribers is returned in a `sub` section of a `{meta}` message.
@ -565,7 +580,10 @@ The `{data}` message represents invites and requests to confirm a subscription.
Message `{get what="info"}` to `me` is automatically replied with a `{meta}` message containing `info` section with the topic parameters (see intro to [Topics][] section). The `public` parameter of `me` topic is associated with the user. Changing it changes `public` not just for the `me` topic, but also everywhere where user's public is shown, such as 'public' of all user's peer to peer topics.
Message `{get what="sub"}` to `me` is different from any other topic as it returns the list of topics that the current user is subscribed to as opposite to the user's subscription to `me`.
Message `{get what="sub"}` to `me` is different from any other topic as it returns the list of topics that the current user is subscribed to as opposite to the user's subscription to `me`. For P2P subscriptions, timestamp of user's last presence and User Agent string are reported:
* seen:
* when: timestamp when the user was last online
* ua: user agent string of the user's client software used last
Message `{get what="data"}` to `me` queries the history of invites/notifications. It's handled the same way as to any other topic.
@ -588,10 +606,14 @@ A group topic is created by sending a `{sub}` message with the topic field set t
A user joining or leaving the topic generates a `{pres}` message to all other users who are currently in the joined state with the topic.
## Support for Client-Side Caching
## Using Server-Issued Message IDs
Tinode provides basic mechanism for implementing client-side caching.
Tinode provides basic support for client-side caching of `{data}` messages in the form of server-issued message IDs. The client may request the last message id from the topic by issuing a `{get what="info"}` message. If the returned ID is greater than the ID of the latest received message, the client knows that the topic has unread messages and their count. The client may fetch these messages using `{get what="data"}` message. The client may also paginate history retrieval by using message IDs.
It's assumed that a single user may connect to the server with multiple client applications. Some of them may be able to cache certain information in client-side storage. An identifier of an instance of the client application is communicated to the server in the `tag` field of the `{login}` message. The `tag` is a string which identifies an instance of the client-side cache. For instance if the user flushes the cache the app should change the application tag. The combination of user ID and tag should be unique, i.e. it's acceptable for different users to have the same tag. If the application does not implement caching it should not provide the `tag`.
## User Agent and Presence Notifications
Server tracks timestamps of user's login, topic join and leave requests keyed by tag. When the user issues a `{get what="data"}` (or an equivalent `{sub get="data"}`), the `browse` may be left empty. Then the server will assume that the `browse.since` is the time when the user left the topic, i.e. "fetch all data since last visit".
A user is reported online when one or more of user's sessions are attached to the `me` topic. Client-side software identifies itself to the server using `ua` (user agent) field of the `{login}` message. The user agent string is published in `{meta}` and `{pres}` messages in the following way:
* When user's first session attaches to `me`, the user agent from that session is broadcast in the `{pres what="on"}` message
* When multiple user sessions are attached to `me`, the user agent of the session where the most recent action has happened is reported in `{pres what="ua"}`; the 'action' in this context means any message sent by the client. To avoid potentially excessive traffic, user agent changes are broadcast no more often than once a minute.
* When user's last session detaches from `me`, the user agent from that session is recorded together with the timestamp; the user agent is broadcast the `{pres what="off"}` message and subsequently reported as the last online timestamp and user agent.

64
server/datamodel.go
View File

@ -159,11 +159,16 @@ type MsgAuthScheme struct {
// Login {login} message
type MsgClientLogin struct {
Id string `json:"id,omitempty"` // Message Id
Scheme string `jdon:"scheme,omitempty"` // Authentication scheme
Secret string `json:"secret"` // Shared secret
ExpireIn JsonDuration `json:"expireIn,omitempty"` // Login expiration time
Tag string `json:"tag,omitempty"` // Device Id
// Message Id
Id string `json:"id,omitempty"`
// User agent
UserAgent string `json:"ua,omitempty"`
// Authentication scheme
Scheme string `jdon:"scheme,omitempty"`
// Shared secret
Secret string `json:"secret"`
// Login expiration time
ExpireIn JsonDuration `json:"expireIn,omitempty"`
}
// Subscription request {sub} message
@ -301,22 +306,20 @@ type ClientComMessage struct {
// Server to client messages
type MsgLastSeenInfo struct {
When time.Time `json:"when"` // when the user was last seen
Tag string `json:"tag,omitempty"` // tag of the device used to access the topic
When *time.Time `json:"when,omitempty"` // when the user was last seen
UserAgent string `json:"ua,omitempty"` // user agent of the device used to access the topic
}
// Topic info, S2C in Meta message
type MsgTopicInfo struct {
CreatedAt *time.Time `json:"created,omitempty"`
UpdatedAt *time.Time `json:"updated,omitempty"`
Name string `json:"name,omitempty"`
DefaultAcs *MsgDefaultAcsMode `json:"defacs,omitempty"`
Acs *MsgAccessMode `json:"acs,omitempty"` // Actual access mode
LastMessage *time.Time `json:"lastMsg,omitempty"` // time of the last {data} message in the topic
LastSeen *MsgLastSeenInfo `json:"seen,omitempty"` // user's last access to topic
LastSeenTag *time.Time `json:"seenTag,omitempty"` // user's last access to topic with the given tag (device)
Public interface{} `json:"public,omitempty"`
Private interface{} `json:"private,omitempty"` // Per-subscription private data
CreatedAt *time.Time `json:"created,omitempty"`
UpdatedAt *time.Time `json:"updated,omitempty"`
Name string `json:"name,omitempty"`
DefaultAcs *MsgDefaultAcsMode `json:"defacs,omitempty"`
Acs *MsgAccessMode `json:"acs,omitempty"` // Actual access mode
SeqId int `json:"seq,omitempty"`
Public interface{} `json:"public,omitempty"`
Private interface{} `json:"private,omitempty"` // Per-subscription private data
}
type MsgAccessMode struct {
@ -331,18 +334,20 @@ type MsgTopicSub struct {
UpdatedAt *time.Time `json:"updated,omitempty"`
Online string `json:"online,omitempty"`
// p2p topics only - id of the other user
With string `json:"with,omitempty"`
// 'me' topic only
LastMsg *time.Time `json:"lastMsg,omitempty"` // last message in a topic, "me' subs only
LastSeen *MsgLastSeenInfo `json:"seen,omitempty"` // user's last access to topic, 'me' subs only
LastSeenTag *time.Time `json:"seenTag,omitempty"` // user's last access to topic with the given tag (device)
// cumulative access mode (mode.Want & mode.Given)
AcsMode string `json:"mode"`
Public interface{} `json:"public,omitempty"`
Private interface{} `json:"private,omitempty"`
// All following makes sence only in context of getting user's subscriptions
// ID of the last {data} message in a topic
SeqId int `json:"seq,omitempty"`
// P2P topics only
// ID of the other user
With string `json:"with,omitempty"`
// Other user's last online timestamp & user agent
LastSeen *MsgLastSeenInfo `json:"seen,omitempty"`
}
type MsgServerCtrl struct {
@ -379,11 +384,10 @@ type MsgServerData struct {
}
type MsgServerPres struct {
Topic string `json:"topic"`
Src string `json:"src"`
What string `json:"what"`
Topic string `json:"topic"`
Src string `json:"src"`
What string `json:"what"`
UserAgent string `json:"ua,omitempty"`
// unroutable, to break the reply loop
isReply bool
}

35
server/db/rethinkdb/adapter.go
View File

@ -272,6 +272,18 @@ func (a *RethinkDbAdapter) UserDelete(appId uint32, id t.Uid, soft bool) error {
return errors.New("UserDelete: not implemented")
}
func (a *RethinkDbAdapter) UserUpdateLastSeen(appid uint32, uid t.Uid, userAgent string, when time.Time) error {
update := struct {
LastSeen time.Time
UserAgent string
}{when, userAgent}
_, err := rdb.DB(a.dbName).Table("users").Get(uid.String()).
Update(update, rdb.UpdateOpts{Durability: "soft"}).RunWrite(a.conn)
return err
}
func (a *RethinkDbAdapter) UserUpdateStatus(appid uint32, uid t.Uid, status interface{}) error {
update := map[string]interface{}{"Status": status}
@ -282,8 +294,7 @@ func (a *RethinkDbAdapter) UserUpdateStatus(appid uint32, uid t.Uid, status inte
}
func (a *RethinkDbAdapter) ChangePassword(appid uint32, id t.Uid, password string) error {
return nil
return errors.New("ChangePassword: not implemented")
}
func (a *RethinkDbAdapter) UserUpdate(appid uint32, uid t.Uid, update map[string]interface{}) error {
@ -404,7 +415,7 @@ func (a *RethinkDbAdapter) TopicsForUser(appid uint32, uid t.Uid) ([]t.Subscript
for rows.Next(&top) {
sub = join[top.Name]
sub.ObjHeader.MergeTimes(&top.ObjHeader)
sub.SetLastMessageAt(top.LastMessageAt)
sub.SetSeqId(top.SeqId)
if strings.HasPrefix(sub.Topic, "grp") {
// all done with a grp topic
sub.SetPublic(top.Public)
@ -433,6 +444,7 @@ func (a *RethinkDbAdapter) TopicsForUser(appid uint32, uid t.Uid) ([]t.Subscript
sub.ObjHeader.MergeTimes(&usr.ObjHeader)
sub.SetWith(uid2.UserId())
sub.SetPublic(usr.Public)
sub.SetLastSeenAndUA(usr.LastSeen, usr.UserAgent)
subs = append(subs, sub)
}
}
@ -509,9 +521,8 @@ func (a *RethinkDbAdapter) TopicDelete(appId uint32, userDbId, topic string) err
func (a *RethinkDbAdapter) TopicUpdateOnMessage(appid uint32, topic string, msg *t.Message) error {
update := struct {
SeqId int
LastMessageAt *time.Time
}{msg.SeqId, &msg.CreatedAt}
SeqId int
}{msg.SeqId}
// Invite - 'me' topic
var err error
@ -528,18 +539,6 @@ func (a *RethinkDbAdapter) TopicUpdateOnMessage(appid uint32, topic string, msg
return err
}
// UpdateLastSeen records the time when a session with a given device ID detached from a topic
func (a *RethinkDbAdapter) UpdateLastSeen(appid uint32, topic string, user t.Uid, tag string, when time.Time) error {
update := struct {
LastSeen map[string]time.Time
}{map[string]time.Time{tag: when}}
_, err := rdb.DB("tinode").Table("subscriptions").Get(topic+":"+user.String()).
Update(update, rdb.UpdateOpts{Durability: "soft"}).RunWrite(a.conn)
return err
}
func (a *RethinkDbAdapter) TopicUpdate(appid uint32, topic string, update map[string]interface{}) error {
_, err := rdb.DB("tinode").Table("topics").GetAllByIndex("Name", topic).Update(update).RunWrite(a.conn)
return err

41
server/hub.go
View File

@ -288,9 +288,6 @@ func topicInit(sreg *sessionJoin, h *Hub) {
t.created = user.CreatedAt
t.updated = user.UpdatedAt
if user.LastMessageAt != nil {
t.lastMessage = *user.LastMessageAt
}
t.lastId = user.SeqId
// Request to create a new p2p topic, then attach to it
@ -379,19 +376,18 @@ func topicInit(sreg *sessionJoin, h *Hub) {
t.created = user1.CreatedAt
t.updated = user1.UpdatedAt
// Newly created, t.lastMessage and t.lastId are not set
// t.lastId is not set (default 0) for new topics
userData.public = user2.GetPublic()
userData.modeWant = user1.ModeWant
userData.modeGiven = user1.ModeGiven
userData.lastSeenTag = user1.LastSeen
t.perUser[userId1] = userData
t.perUser[userId2] = perUserData{
public: user1.GetPublic(),
modeWant: user2.ModeWant,
modeGiven: user2.ModeGiven,
lastSeenTag: user2.LastSeen}
public: user1.GetPublic(),
modeWant: user2.ModeWant,
modeGiven: user2.ModeGiven,
}
t.original = t.name
sreg.created = true
@ -436,20 +432,17 @@ func topicInit(sreg *sessionJoin, h *Hub) {
t.created = stopic.CreatedAt
t.updated = stopic.UpdatedAt
if stopic.LastMessageAt != nil {
t.lastMessage = *stopic.LastMessageAt
}
t.lastId = stopic.SeqId
for i := 0; i < 2; i++ {
uid := types.ParseUid(subs[i].User)
t.perUser[uid] = perUserData{
// Based on other user
public: subs[(i+1)%2].GetPublic(),
private: subs[i].Private,
lastSeenTag: subs[i].LastSeen,
modeWant: subs[i].ModeWant,
modeGiven: subs[i].ModeGiven}
public: subs[(i+1)%2].GetPublic(),
private: subs[i].Private,
// lastSeenTag: subs[i].LastSeen,
modeWant: subs[i].ModeWant,
modeGiven: subs[i].ModeGiven}
}
// Processing request to create a new generic (group) topic:
@ -502,7 +495,7 @@ func topicInit(sreg *sessionJoin, h *Hub) {
t.created = timestamp
t.updated = timestamp
// t.lastId and t.lastMessage are not set for new topics
// t.lastId is not set for new topics
stopic := &types.Topic{
ObjHeader: types.ObjHeader{CreatedAt: timestamp},
@ -555,9 +548,6 @@ func topicInit(sreg *sessionJoin, h *Hub) {
t.created = stopic.CreatedAt
t.updated = stopic.UpdatedAt
if stopic.LastMessageAt != nil {
t.lastMessage = *stopic.LastMessageAt
}
t.lastId = stopic.SeqId
}
@ -582,10 +572,10 @@ func (t *Topic) loadSubscribers() error {
for _, sub := range subs {
uid := types.ParseUid(sub.User)
t.perUser[uid] = perUserData{
private: sub.Private,
lastSeenTag: sub.LastSeen, // could be nil
modeWant: sub.ModeWant,
modeGiven: sub.ModeGiven}
private: sub.Private,
//lastSeenTag: sub.LastSeen, // could be nil
modeWant: sub.ModeWant,
modeGiven: sub.ModeGiven}
if sub.ModeGiven&sub.ModeWant&types.ModeOwner != 0 {
log.Printf("hub.loadSubscriptions: %s set owner to %s", t.name, uid.String())
@ -607,7 +597,6 @@ func replyTopicInfoBasic(sess *Session, topic string, get *MsgClientGet) {
if err == nil {
info.CreatedAt = &stopic.CreatedAt
info.UpdatedAt = &stopic.UpdatedAt
info.LastMessage = stopic.LastMessageAt
info.Public = stopic.Public
} else {
simpleByteSender(sess.send, ErrUnknown(get.Id, get.Topic, now))

4
server/pres.go
View File

@ -103,9 +103,9 @@ func (t *Topic) loadContacts(uid types.Uid) error {
// Me topic activated, deactivated or updated, push presence to contacts
// Case 1.a.iii, 2, 3
func (t *Topic) presPubMeChange(what string) {
func (t *Topic) presPubMeChange(what string, ua string) {
// Push update to subscriptions
update := &MsgServerPres{Topic: "me", What: what, Src: t.name}
update := &MsgServerPres{Topic: "me", What: what, Src: t.name, UserAgent: ua}
for topic, _ := range t.perSubs {
globals.hub.route <- &ServerComMessage{Pres: update, appid: t.appid, rcptto: topic}
//log.Printf("Pres 1.a.ii, 2, 3: from '%s' (src: %s) to %s [%s]", t.name, update.Src, topic, what)

17
server/session.go
View File

@ -50,8 +50,6 @@ const (
NONE = iota
WEBSOCK
LPOLL
TAG_UNDEF = "-"
)
/*
@ -74,15 +72,18 @@ type Session struct {
// Application ID
appid uint32
// Client instance tag, a string provived by an authenticated client for managing client-side cache
tag string
// User agent, a string provived by an authenticated client
ua string
// ID of the current user or 0
uid types.Uid
// Time when the session was last touched
// Time when the long polling session was last refreshed
lastTouched time.Time
// Time when the session received any packer from client
lastAction time.Time
// outbound mesages, buffered
send chan []byte
@ -153,6 +154,7 @@ func (s *Session) dispatch(raw []byte) {
log.Printf("Session.dispatch got '%s' from '%s'", raw, s.remoteAddr)
timestamp := time.Now().UTC().Round(time.Millisecond)
s.lastAction = timestamp
if err := json.Unmarshal(raw, &msg); err != nil {
// Malformed message
log.Println("Session.dispatch: " + err.Error())
@ -340,10 +342,7 @@ func (s *Session) login(msg *ClientComMessage) {
}
s.uid = uid
s.tag = msg.Login.Tag
if s.tag == "" {
s.tag = TAG_UNDEF
}
s.ua = msg.Login.UserAgent
expireIn := time.Duration(msg.Login.ExpireIn)
if expireIn <= 0 || expireIn > TOKEN_LIFETIME_MAX {

7
server/store/adapter/adapter.go
View File

@ -20,25 +20,26 @@ type Adapter interface {
UserCreate(appid uint32, usr *t.User) (err error, dupeUserName bool)
UserGet(appId uint32, id t.Uid) (*t.User, error)
UserGetAll(appId uint32, ids ...t.Uid) ([]t.User, error)
//GetLastSeenAndStatus(appid uint32, id t.Uid) (time.Time, interface{}, error)
UserFind(appId uint32, params map[string]interface{}) ([]t.User, error)
UserDelete(appId uint32, id t.Uid, soft bool) error
UserUpdateLastSeen(appid uint32, uid t.Uid, userAgent string, when time.Time) error
UserUpdateStatus(appid uint32, uid t.Uid, status interface{}) error
ChangePassword(appid uint32, id t.Uid, password string) error
UserUpdate(appid uint32, uid t.Uid, update map[string]interface{}) error
// Topic/contact management
// TopicCreate creates a topic
TopicCreate(appid uint32, topic *t.Topic) error
// TopicCreateP2P creates a p2p topic
TopicCreateP2P(appId uint32, initiator, invited *t.Subscription) error
// TopicGet loads a single topic by name, if it exists. If the topic does not exist the call returns (nil, nil)
TopicGet(appid uint32, topic string) (*t.Topic, error)
// TopicsForUser loads subscriptions for a given user
TopicsForUser(appid uint32, uid t.Uid) ([]t.Subscription, error)
// UsersForTopic loads users' subscriptions for a given topic
UsersForTopic(appid uint32, topic string) ([]t.Subscription, error)
//UsersForP2P(appid uint32, uid1, uid2 t.Uid) ([]t.Subscription, error)
TopicShare(appid uint32, acl []t.Subscription) (int, error)
UpdateLastSeen(appid uint32, topic string, uid t.Uid, tag string, when time.Time) error
TopicDelete(appid uint32, userDbId, topic string) error
TopicUpdateOnMessage(appid uint32, topic string, msg *t.Message) error
TopicUpdate(appid uint32, topic string, update map[string]interface{}) error

14
server/store/store.go
View File

@ -187,12 +187,6 @@ func (UsersObjMapper) GetAll(appid uint32, uid ...types.Uid) ([]types.User, erro
return adaptr.UserGetAll(appid, uid...)
}
/*
func (u UsersObjMapper) GetLastSeenAndStatus(appid uint32, id types.Uid) (time.Time, interface{}, error) {
return adaptr.GetLastSeenAndStatus(appid, id)
}
*/
// TODO(gene): implement
func (UsersObjMapper) Find(appId uint32, params map[string]interface{}) ([]types.User, error) {
return nil, errors.New("store: not implemented")
@ -207,6 +201,10 @@ func (UsersObjMapper) UpdateStatus(appid uint32, id types.Uid, status interface{
return errors.New("store: not implemented")
}
func (UsersObjMapper) UpdateLastSeen(appid uint32, uid types.Uid, userAgent string, when time.Time) error {
return adaptr.UserUpdateLastSeen(appid, uid, userAgent, when)
}
// ChangePassword changes user's password in "basic" authentication scheme
func (UsersObjMapper) ChangeAuthCredential(appid uint32, uid types.Uid, scheme, secret string) error {
if scheme == "basic" {
@ -285,10 +283,6 @@ func (TopicsObjMapper) GetSubs(appid uint32, topic string) ([]types.Subscription
return adaptr.SubsForTopic(appid, topic)
}
func (TopicsObjMapper) UpdateLastSeen(appid uint32, topic string, id types.Uid, tag string, when time.Time) error {
return adaptr.UpdateLastSeen(appid, topic, id, tag, when)
}
func (TopicsObjMapper) Update(appid uint32, topic string, update map[string]interface{}) error {
update["UpdatedAt"] = types.TimeNow()
return adaptr.TopicUpdate(appid, topic, update)

39
server/store/types/types.go
View File

@ -236,8 +236,10 @@ type User struct {
// Values for 'me' topic:
// Server-issued sequence ID for messages in 'me'
SeqId int
// Deprecated
LastMessageAt *time.Time
// Last time when the user joined 'me' topic, by User Agent
LastSeen time.Time
// User agent provided when accessing the topic last time
UserAgent string
Public interface{}
}
@ -404,14 +406,14 @@ type Subscription struct {
ModeGiven AccessMode // Granted access
ClearedAt *time.Time // User deleted messages older than this time; TODO(gene): topic owner can hard-delete messages
LastSeen map[string]time.Time // Last time when the user joined the topic, by device tag
Private interface{} // User's private data associated with the subscription to topic
// Deserialized ephemeral values
public interface{} // Deserialized public value from topic or user (depends on context)
with string // p2p topics only: id of the other user
lastMessageAt *time.Time
public interface{} // Deserialized public value from topic or user (depends on context)
with string // p2p topics only: id of the other user
seqId int // deserialized SeqID from user or topic
lastSeen time.Time // timestamp when the user was last online
userAgent string // user agent string of the last online access
}
// SetPublic assigns to public, otherwise not accessible from outside the package
@ -431,12 +433,25 @@ func (s *Subscription) GetWith() string {
return s.with
}
func (s *Subscription) GetLastMessageAt() *time.Time {
return s.lastMessageAt
func (s *Subscription) GetSeqId() int {
return s.seqId
}
func (s *Subscription) SetLastMessageAt(lm *time.Time) {
s.lastMessageAt = lm
func (s *Subscription) SetSeqId(id int) {
s.seqId = id
}
func (s *Subscription) GetLastSeen() time.Time {
return s.lastSeen
}
func (s *Subscription) GetUserAgent() string {
return s.userAgent
}
func (s *Subscription) SetLastSeenAndUA(when time.Time, ua string) {
s.lastSeen = when
s.userAgent = ua
}
type perUserData struct {
@ -457,8 +472,6 @@ type Topic struct {
// Default access to topic
Access DefaultAccess
// Deprecated
LastMessageAt *time.Time
// Server-issued sequential ID
SeqId int

88
server/topic.go
View File

@ -60,9 +60,6 @@ type Topic struct {
// Time when the topic was last updated
updated time.Time
// Time of the last message
lastMessage time.Time
// Server-side ID of the last data message
lastId int
@ -112,8 +109,8 @@ const (
type perUserData struct {
online int
private interface{}
lastSeenTag map[string]time.Time
private interface{}
// lastSeenTag map[string]time.Time
// cleared time.Time // time, when the topic was cleared by the user
modeWant types.AccessMode
modeGiven types.AccessMode
@ -178,18 +175,16 @@ func (t *Topic) run(hub *Hub) {
delete(t.sessions, leave.sess)
pud := t.perUser[leave.sess.uid]
if pud.lastSeenTag == nil {
pud.lastSeenTag = map[string]time.Time{}
}
pud.lastSeenTag[leave.sess.tag] = now
pud.online--
t.perUser[leave.sess.uid] = pud
if t.cat == TopicCat_Grp && pud.online == 0 {
if t.cat == TopicCat_Me {
// Update user's last online timestamp & user agent
if err := store.Users.UpdateLastSeen(t.appid, leave.sess.uid, leave.sess.ua, now); err != nil {
log.Println(err)
}
} else if t.cat == TopicCat_Grp && pud.online == 0 {
t.presPubChange(leave.sess.uid.UserId(), "off")
}
if err := store.Topics.UpdateLastSeen(t.appid, t.name, leave.sess.uid, leave.sess.tag, now); err != nil {
log.Println(err)
}
}
// If there are no more subscriptions to this topic, start a kill timer
@ -231,7 +226,6 @@ func (t *Topic) run(hub *Hub) {
continue
}
t.lastMessage = msg.timestamp
t.lastId++
msg.Data.SeqId = t.lastId
@ -301,7 +295,7 @@ func (t *Topic) run(hub *Hub) {
hub.unreg <- topicUnreg{appid: t.appid, topic: t.name}
// FIXME(gene): save lastMessage value;
if t.cat == TopicCat_Me {
t.presPubMeChange("off")
t.presPubMeChange("off", "")
} else if t.cat == TopicCat_Grp {
t.presPubTopicOnline(false)
} // not publishing online/offline to P2P topics
@ -349,7 +343,7 @@ func (t *Topic) handleSubscription(h *Hub, sreg *sessionJoin) error {
log.Printf("hub: failed to load contacts for '%s'", t.name)
}
t.presPubMeChange("on")
t.presPubMeChange("on", sreg.sess.ua)
} else if t.cat == TopicCat_Grp {
t.presPubTopicOnline(true)
@ -461,10 +455,10 @@ func (t *Topic) requestSub(h *Hub, sess *Session, pktId string, want string, inf
}
userData = perUserData{
private: private,
modeGiven: t.accessAuth,
modeWant: modeWant,
lastSeenTag: make(map[string]time.Time),
private: private,
modeGiven: t.accessAuth,
modeWant: modeWant,
//lastSeenTag: make(map[string]time.Time),
}
// Add subscription to database
@ -473,7 +467,6 @@ func (t *Topic) requestSub(h *Hub, sess *Session, pktId string, want string, inf
Topic: t.name,
ModeWant: userData.modeWant,
ModeGiven: userData.modeGiven,
LastSeen: userData.lastSeenTag,
Private: userData.private,
}
@ -657,7 +650,6 @@ func (t *Topic) approveSub(h *Hub, sess *Session, target types.Uid, set *MsgClie
Topic: t.name,
ModeWant: types.ModeNone,
ModeGiven: modeGiven,
LastSeen: make(map[string]time.Time),
}
if err := store.Subs.Create(t.appid, &sub); err != nil {
@ -666,10 +658,9 @@ func (t *Topic) approveSub(h *Hub, sess *Session, target types.Uid, set *MsgClie
}
userData = perUserData{
modeGiven: sub.ModeGiven,
modeWant: sub.ModeWant,
private: nil,
lastSeenTag: sub.LastSeen,
modeGiven: sub.ModeGiven,
modeWant: sub.ModeWant,
private: nil,
}
t.perUser[target] = userData
@ -743,7 +734,8 @@ func (t *Topic) replyGetInfo(sess *Session, id string, created bool) error {
info.Public = pud.public
}
// Request may come from a subscriber or a stranger. Give a stranger a lot less info than a subscriber
// Request may come from a subscriber (full == true) or a stranger.
// Give subscriber a lot more info than to a stranger
if full {
if pud.modeGiven&pud.modeWant&types.ModeShare != 0 {
info.DefaultAcs = &MsgDefaultAcsMode{
@ -757,20 +749,7 @@ func (t *Topic) replyGetInfo(sess *Session, id string, created bool) error {
info.Private = pud.private
if !t.lastMessage.IsZero() {
info.LastMessage = &t.lastMessage
}
if when, ok := pud.lastSeenTag[sess.tag]; ok {
info.LastSeenTag = &when
}
if len(pud.lastSeenTag) > 0 {
tag, when := mostRecent(pud.lastSeenTag)
if !when.IsZero() {
info.LastSeen = &MsgLastSeenInfo{When: when, Tag: tag}
}
}
info.SeqId = t.lastId
// When a topic is first created, its name may change. Report the new name
if created {
@ -936,17 +915,14 @@ func (t *Topic) replyGetSub(sess *Session, id string) error {
uid := types.ParseUid(sub.User)
if t.cat == TopicCat_Me {
mts.Topic = sub.Topic
mts.SeqId = sub.GetSeqId()
mts.With = sub.GetWith()
mts.LastMsg = sub.GetLastMessageAt()
mts.UpdatedAt = &sub.UpdatedAt
if when, ok := sub.LastSeen[sess.tag]; ok && !when.IsZero() {
mts.LastSeenTag = &when
}
if len(sub.LastSeen) > 0 {
tag, when := mostRecent(sub.LastSeen)
if !when.IsZero() {
mts.LastSeen = &MsgLastSeenInfo{When: when, Tag: tag}
}
lastSeen := sub.GetLastSeen()
if !lastSeen.IsZero() {
mts.LastSeen = &MsgLastSeenInfo{
When: &lastSeen,
UserAgent: sub.GetUserAgent()}
}
} else {
mts.User = uid.UserId()
@ -1093,6 +1069,18 @@ func (t *Topic) evictUser(uid types.Uid, clear bool, ignore *Session) {
}
}
func (t *Topic) mostRecentSession() *Session {
var sess *Session
var latest time.Time
for s, _ := range t.sessions {
if s.lastAction.After(latest) {
sess = s
latest = s.lastAction
}
}
return sess
}
func msgOpts2storeOpts(req *MsgBrowseOpts) *types.BrowseOpt {
var opts *types.BrowseOpt
if req != nil {

2
tinode-db/makedb.go
View File

@ -39,7 +39,7 @@ func main() {
var reset = flag.Bool("reset", false, "first delete the database if one exists")
var datafile = flag.String("data", "", "path to sample data to load")
var conffile = flag.String("config", "./config", "config of the database connection")
var conffile = flag.String("config", "./tinode.conf", "config of the database connection")
flag.Parse()
var data Data

17
tinode-db/rethink.go
View File

@ -202,7 +202,9 @@ func gen_rethink(reset bool, dbsource string, data *Data) {
log.Println("Generating messages...")
rand.Seed(time.Now().UnixNano())
seqId := 0
seqIds := map[string]int{}
// Starting 4 days ago
timestamp := time.Now().UTC().Round(time.Millisecond).Add(time.Second * time.Duration(-3600*24*4))
for i := 0; i < 80; i++ {
sub := data.Subscriptions[rand.Intn(len(data.Subscriptions))]
@ -213,18 +215,21 @@ func gen_rethink(reset bool, dbsource string, data *Data) {
topic = uid.P2PName(from)
}
seqId++
seqIds[topic]++
seqId := seqIds[topic]
str := data.Messages[rand.Intn(len(data.Messages))]
timestamp := time.Now().UTC().Round(time.Millisecond).Add(time.Second * time.Duration(-1*rand.Intn(3600*24*4))) // spread over the past 4 days
if err = store.Messages.Save(0, &types.Message{
// Max time between messages is 2 hours, averate - 1 hour, time is increasing as seqId increases
timestamp = timestamp.Add(time.Second * time.Duration(rand.Intn(3600*2)))
msg := types.Message{
ObjHeader: types.ObjHeader{CreatedAt: timestamp},
SeqId: seqId,
Topic: topic,
From: from.String(),
Content: str}); err != nil {
Content: str}
if err = store.Messages.Save(0, &msg); err != nil {
log.Fatal(err)
}
log.Printf("Message from '%s' to '%s' (%s) '%s'", from.String(), topic, nameIndex[sub["topic"].(string)], str)
log.Printf("Message %d at %v to '%s' '%s'", msg.SeqId, msg.CreatedAt, topic, str)
}
}