diff --git a/resource/entry.go b/resource/entry.go index 37d9ce4..9a77700 100644 --- a/resource/entry.go +++ b/resource/entry.go @@ -143,7 +143,7 @@ func AddEntry(feedEntry Entry, FeedId string, ApplicationId string, OrgId string // notify data, _ := json.Marshal(entry) - room.Publish <- room.NewEvent(model.EVENT_MESSAGE, "system", string(data)) + room.Publish <- room.NewEntryEvent(model.EVENT_MESSAGE, "system", string(data)) return feedEntry.Id, nil } diff --git a/resource/feed.go b/resource/feed.go index eb3ce9b..c8a735a 100644 --- a/resource/feed.go +++ b/resource/feed.go @@ -95,7 +95,7 @@ func AddFeed(feed Feed, applicationId string, orgId string) (id string, err erro // notify data, _ := json.Marshal(feed) - room.Publish <- room.NewEvent(model.EVENT_MESSAGE, "system", string(data)) + room.Publish <- room.NewFeedEvent(model.EVENT_MESSAGE, "system", string(data)) return feed.Id, nil } diff --git a/service/api/api.go b/service/api/api.go deleted file mode 100644 index f9ce540..0000000 --- a/service/api/api.go +++ /dev/null @@ -1,7 +0,0 @@ -package api - -import ( - _ "github.com/feedlabs/elasticfeed/service/api/v1" -) - -func init() {} diff --git a/service/api/v1/api.go b/service/api/v1/api.go deleted file mode 100644 index 4bf8636..0000000 --- a/service/api/v1/api.go +++ /dev/null @@ -1,8 +0,0 @@ -package v1 - -import ( - _ "github.com/feedlabs/elasticfeed/service/api/v1/router" - _ "github.com/feedlabs/elasticfeed/service/api/v1/controller" -) - -func init() {} diff --git a/service/db/api.go b/service/db/api.go new file mode 100644 index 0000000..c51e66e --- /dev/null +++ b/service/db/api.go @@ -0,0 +1,7 @@ +package api + +import ( + _ "github.com/feedlabs/elasticfeed/service/db/v1" +) + +func init() {} diff --git a/service/db/v1/api.go b/service/db/v1/api.go new file mode 100644 index 0000000..522b251 --- /dev/null +++ b/service/db/v1/api.go @@ -0,0 +1,8 @@ +package v1 + +import ( + _ "github.com/feedlabs/elasticfeed/service/db/v1/router" + _ "github.com/feedlabs/elasticfeed/service/db/v1/controller" +) + +func init() {} diff --git a/service/api/v1/controller/admin.go b/service/db/v1/controller/admin.go similarity index 97% rename from service/api/v1/controller/admin.go rename to service/db/v1/controller/admin.go index 2c8774b..71807de 100644 --- a/service/api/v1/controller/admin.go +++ b/service/db/v1/controller/admin.go @@ -4,7 +4,7 @@ import ( "encoding/json" "github.com/feedlabs/elasticfeed/resource" - "github.com/feedlabs/elasticfeed/service/api/v1/template/admin" + "github.com/feedlabs/elasticfeed/service/db/v1/template/admin" ) type AdminController struct { diff --git a/service/api/v1/controller/application.go b/service/db/v1/controller/application.go similarity index 98% rename from service/api/v1/controller/application.go rename to service/db/v1/controller/application.go index 61fe698..63c1bd0 100644 --- a/service/api/v1/controller/application.go +++ b/service/db/v1/controller/application.go @@ -4,7 +4,7 @@ import ( "encoding/json" "github.com/feedlabs/elasticfeed/resource" - "github.com/feedlabs/elasticfeed/service/api/v1/template/application" + "github.com/feedlabs/elasticfeed/service/db/v1/template/application" ) type ApplicationController struct { diff --git a/service/api/v1/controller/default.go b/service/db/v1/controller/default.go similarity index 95% rename from service/api/v1/controller/default.go rename to service/db/v1/controller/default.go index 3c3e296..63a464a 100644 --- a/service/api/v1/controller/default.go +++ b/service/db/v1/controller/default.go @@ -7,7 +7,7 @@ import ( "github.com/feedlabs/feedify" "github.com/feedlabs/elasticfeed/resource" "github.com/feedlabs/elasticfeed/helper" - "github.com/feedlabs/elasticfeed/service/api/v1/template" + "github.com/feedlabs/elasticfeed/service/db/v1/template" ) type DefaultController struct { @@ -70,5 +70,5 @@ func NoRoutes() { func init() { SetAuthenticationFilter() SetGlobalResponseHeader() -// NoRoutes() + // NoRoutes() } diff --git a/service/api/v1/controller/entry.go b/service/db/v1/controller/entry.go similarity index 98% rename from service/api/v1/controller/entry.go rename to service/db/v1/controller/entry.go index f0258c7..8bc6a24 100644 --- a/service/api/v1/controller/entry.go +++ b/service/db/v1/controller/entry.go @@ -4,7 +4,7 @@ import ( "encoding/json" "github.com/feedlabs/elasticfeed/resource" - "github.com/feedlabs/elasticfeed/service/api/v1/template/entry" + "github.com/feedlabs/elasticfeed/service/db/v1/template/entry" ) type EntryController struct { diff --git a/service/api/v1/controller/feed.go b/service/db/v1/controller/feed.go similarity index 98% rename from service/api/v1/controller/feed.go rename to service/db/v1/controller/feed.go index c207266..e9a4b29 100644 --- a/service/api/v1/controller/feed.go +++ b/service/db/v1/controller/feed.go @@ -4,7 +4,7 @@ import ( "encoding/json" "github.com/feedlabs/elasticfeed/resource" - "github.com/feedlabs/elasticfeed/service/api/v1/template/feed" + "github.com/feedlabs/elasticfeed/service/db/v1/template/feed" ) type FeedController struct { diff --git a/service/api/v1/controller/org.go b/service/db/v1/controller/org.go similarity index 97% rename from service/api/v1/controller/org.go rename to service/db/v1/controller/org.go index f62d96a..c26211a 100644 --- a/service/api/v1/controller/org.go +++ b/service/db/v1/controller/org.go @@ -4,7 +4,7 @@ import ( "encoding/json" "github.com/feedlabs/elasticfeed/resource" - template "github.com/feedlabs/elasticfeed/service/api/v1/template/org" + template "github.com/feedlabs/elasticfeed/service/db/v1/template/org" ) type OrgController struct { diff --git a/service/api/v1/controller/token.go b/service/db/v1/controller/token.go similarity index 98% rename from service/api/v1/controller/token.go rename to service/db/v1/controller/token.go index cdf5427..4b7cad6 100644 --- a/service/api/v1/controller/token.go +++ b/service/db/v1/controller/token.go @@ -4,7 +4,7 @@ import ( "encoding/json" "github.com/feedlabs/elasticfeed/resource" - "github.com/feedlabs/elasticfeed/service/api/v1/template/token" + "github.com/feedlabs/elasticfeed/service/db/v1/template/token" ) type TokenController struct { diff --git a/service/api/v1/router/admin.go b/service/db/v1/router/admin.go similarity index 81% rename from service/api/v1/router/admin.go rename to service/db/v1/router/admin.go index 7d8ccc8..47bfbd8 100644 --- a/service/api/v1/router/admin.go +++ b/service/db/v1/router/admin.go @@ -2,7 +2,7 @@ package router import ( "github.com/feedlabs/feedify" - "github.com/feedlabs/elasticfeed/service/api/v1/controller" + "github.com/feedlabs/elasticfeed/service/db/v1/controller" ) func init() { diff --git a/service/api/v1/router/application.go b/service/db/v1/router/application.go similarity index 83% rename from service/api/v1/router/application.go rename to service/db/v1/router/application.go index f16be7c..4a72333 100644 --- a/service/api/v1/router/application.go +++ b/service/db/v1/router/application.go @@ -2,7 +2,7 @@ package router import ( "github.com/feedlabs/feedify" - "github.com/feedlabs/elasticfeed/service/api/v1/controller" + "github.com/feedlabs/elasticfeed/service/db/v1/controller" ) func init() { diff --git a/service/api/v1/router/default.go b/service/db/v1/router/default.go similarity index 77% rename from service/api/v1/router/default.go rename to service/db/v1/router/default.go index 95ad3aa..82db7f3 100644 --- a/service/api/v1/router/default.go +++ b/service/db/v1/router/default.go @@ -2,7 +2,7 @@ package router import ( "github.com/feedlabs/feedify" - "github.com/feedlabs/elasticfeed/service/api/v1/controller" + "github.com/feedlabs/elasticfeed/service/db/v1/controller" ) func init() { diff --git a/service/api/v1/router/entry.go b/service/db/v1/router/entry.go similarity index 91% rename from service/api/v1/router/entry.go rename to service/db/v1/router/entry.go index 2ed16f6..4adad9f 100644 --- a/service/api/v1/router/entry.go +++ b/service/db/v1/router/entry.go @@ -2,7 +2,7 @@ package router import ( "github.com/feedlabs/feedify" - "github.com/feedlabs/elasticfeed/service/api/v1/controller" + "github.com/feedlabs/elasticfeed/service/db/v1/controller" ) func init() { diff --git a/service/api/v1/router/feed.go b/service/db/v1/router/feed.go similarity index 84% rename from service/api/v1/router/feed.go rename to service/db/v1/router/feed.go index 9e171ed..bf70232 100644 --- a/service/api/v1/router/feed.go +++ b/service/db/v1/router/feed.go @@ -2,7 +2,7 @@ package router import ( "github.com/feedlabs/feedify" - "github.com/feedlabs/elasticfeed/service/api/v1/controller" + "github.com/feedlabs/elasticfeed/service/db/v1/controller" ) func init() { diff --git a/service/api/v1/router/org.go b/service/db/v1/router/org.go similarity index 80% rename from service/api/v1/router/org.go rename to service/db/v1/router/org.go index 7b0064a..6a6b08a 100644 --- a/service/api/v1/router/org.go +++ b/service/db/v1/router/org.go @@ -2,7 +2,7 @@ package router import ( "github.com/feedlabs/feedify" - "github.com/feedlabs/elasticfeed/service/api/v1/controller" + "github.com/feedlabs/elasticfeed/service/db/v1/controller" ) func init() { diff --git a/service/api/v1/router/token.go b/service/db/v1/router/token.go similarity index 90% rename from service/api/v1/router/token.go rename to service/db/v1/router/token.go index ff3a90c..f3be790 100644 --- a/service/api/v1/router/token.go +++ b/service/db/v1/router/token.go @@ -2,7 +2,7 @@ package router import ( "github.com/feedlabs/feedify" - "github.com/feedlabs/elasticfeed/service/api/v1/controller" + "github.com/feedlabs/elasticfeed/service/db/v1/controller" ) func init() { diff --git a/service/api/v1/template/admin/request.go b/service/db/v1/template/admin/request.go similarity index 100% rename from service/api/v1/template/admin/request.go rename to service/db/v1/template/admin/request.go diff --git a/service/api/v1/template/admin/response.go b/service/db/v1/template/admin/response.go similarity index 100% rename from service/api/v1/template/admin/response.go rename to service/db/v1/template/admin/response.go diff --git a/service/api/v1/template/application/request.go b/service/db/v1/template/application/request.go similarity index 100% rename from service/api/v1/template/application/request.go rename to service/db/v1/template/application/request.go diff --git a/service/api/v1/template/application/response.go b/service/db/v1/template/application/response.go similarity index 100% rename from service/api/v1/template/application/response.go rename to service/db/v1/template/application/response.go diff --git a/service/api/v1/template/const.go b/service/db/v1/template/const.go similarity index 100% rename from service/api/v1/template/const.go rename to service/db/v1/template/const.go diff --git a/service/api/v1/template/default.go b/service/db/v1/template/default.go similarity index 100% rename from service/api/v1/template/default.go rename to service/db/v1/template/default.go diff --git a/service/api/v1/template/entry/request.go b/service/db/v1/template/entry/request.go similarity index 100% rename from service/api/v1/template/entry/request.go rename to service/db/v1/template/entry/request.go diff --git a/service/api/v1/template/entry/response.go b/service/db/v1/template/entry/response.go similarity index 100% rename from service/api/v1/template/entry/response.go rename to service/db/v1/template/entry/response.go diff --git a/service/api/v1/template/feed/request.go b/service/db/v1/template/feed/request.go similarity index 100% rename from service/api/v1/template/feed/request.go rename to service/db/v1/template/feed/request.go diff --git a/service/api/v1/template/feed/response.go b/service/db/v1/template/feed/response.go similarity index 100% rename from service/api/v1/template/feed/response.go rename to service/db/v1/template/feed/response.go diff --git a/service/api/v1/template/org/request.go b/service/db/v1/template/org/request.go similarity index 96% rename from service/api/v1/template/org/request.go rename to service/db/v1/template/org/request.go index 0fd6a0e..0efc62b 100644 --- a/service/api/v1/template/org/request.go +++ b/service/db/v1/template/org/request.go @@ -3,7 +3,7 @@ package org import ( "errors" "github.com/feedlabs/feedify/context" - "github.com/feedlabs/elasticfeed/service/api/v1/template" + "github.com/feedlabs/elasticfeed/service/db/v1/template" ) func CheckRequiredParams() { diff --git a/service/api/v1/template/org/response.go b/service/db/v1/template/org/response.go similarity index 98% rename from service/api/v1/template/org/response.go rename to service/db/v1/template/org/response.go index 07969d8..3c951bf 100644 --- a/service/api/v1/template/org/response.go +++ b/service/db/v1/template/org/response.go @@ -3,7 +3,7 @@ package org import ( "strconv" "github.com/feedlabs/elasticfeed/resource" - "github.com/feedlabs/elasticfeed/service/api/v1/template" + "github.com/feedlabs/elasticfeed/service/db/v1/template" "errors" "sort" ) diff --git a/service/api/v1/template/request.go b/service/db/v1/template/request.go similarity index 100% rename from service/api/v1/template/request.go rename to service/db/v1/template/request.go diff --git a/service/api/v1/template/response.go b/service/db/v1/template/response.go similarity index 100% rename from service/api/v1/template/response.go rename to service/db/v1/template/response.go diff --git a/service/api/v1/template/token/request.go b/service/db/v1/template/token/request.go similarity index 100% rename from service/api/v1/template/token/request.go rename to service/db/v1/template/token/request.go diff --git a/service/api/v1/template/token/response.go b/service/db/v1/template/token/response.go similarity index 100% rename from service/api/v1/template/token/response.go rename to service/db/v1/template/token/response.go diff --git a/service/service.go b/service/service.go index 64fa20c..3e3d97e 100644 --- a/service/service.go +++ b/service/service.go @@ -1,8 +1,9 @@ package service import ( - _ "github.com/feedlabs/elasticfeed/service/api" + _ "github.com/feedlabs/elasticfeed/service/db" _ "github.com/feedlabs/elasticfeed/service/stream" + _ "github.com/feedlabs/elasticfeed/service/system" ) func init() {} diff --git a/service/stream/controller/channel/long_pooling.go b/service/stream/controller/channel/long_pooling.go index efcf595..67e5380 100644 --- a/service/stream/controller/channel/long_pooling.go +++ b/service/stream/controller/channel/long_pooling.go @@ -12,22 +12,32 @@ type LongPollingController struct { } func (this *LongPollingController) Join() { - uname := this.GetString("uname") - if len(uname) == 0 { + chid := this.GetString("chid") + if len(chid) == 0 { return } - room.Join(uname, nil) + w := this.GetCtx().ResponseWriter + r := this.GetCtx().Input.Request + sess := room.GlobalSessions.SessionStart(w, r) + defer sess.SessionRelease(w) + + room.Join(chid, nil) } func (this *LongPollingController) Post() { - uname := this.GetString("uname") - content := this.GetString("content") - if len(uname) == 0 || len(content) == 0 { + chid := this.GetString("chid") + data := this.GetString("data") + if len(chid) == 0 || len(data) == 0 { return } - room.Publish <- room.NewEvent(model.EVENT_MESSAGE, uname, content) + // Feature: + // or specific request for this client; + // should be executed and returned directly to user + // lastReceived time should not be changed in that case + + room.Publish <- room.NewSystemEvent(model.EVENT_MESSAGE, chid, data) } func (this *LongPollingController) Fetch() { diff --git a/service/stream/controller/channel/websocket.go b/service/stream/controller/channel/websocket.go index 11282ce..ea8619d 100644 --- a/service/stream/controller/channel/websocket.go +++ b/service/stream/controller/channel/websocket.go @@ -15,12 +15,17 @@ type WebSocketController struct { } func (this *WebSocketController) Join() { - uname := this.GetString("uname") - if len(uname) == 0 { + chid := this.GetString("chid") + if len(chid) == 0 { return } - ws, err := websocket.Upgrade(this.Ctx.ResponseWriter, this.Ctx.Request, nil, 1024, 1024) + w := this.GetCtx().ResponseWriter + r := this.GetCtx().Input.Request + sess := room.GlobalSessions.SessionStart(w, r) + defer sess.SessionRelease(w) + + ws, err := websocket.Upgrade(w, r, nil, 1024, 1024) if _, ok := err.(websocket.HandshakeError); ok { http.Error(this.Ctx.ResponseWriter, "Not a websocket handshake", 400) return @@ -29,16 +34,16 @@ func (this *WebSocketController) Join() { return } - room.Join(uname, ws) - defer room.Leave(uname) + room.Join(chid, ws) + defer room.Leave(chid) for { _, p, err := ws.ReadMessage() if err != nil { return } - room.Publish <- room.NewEvent(model.EVENT_MESSAGE, uname, string(p)) - room.System_rpc <- ws + room.Publish <- room.NewSystemEvent(model.EVENT_MESSAGE, chid, string(p)) + room.P2P <- ws } } diff --git a/service/stream/controller/room/feed.go b/service/stream/controller/room/feed.go index 7c3bc16..dc1fbdd 100644 --- a/service/stream/controller/room/feed.go +++ b/service/stream/controller/room/feed.go @@ -8,41 +8,73 @@ import ( "github.com/feedlabs/feedify" "github.com/gorilla/websocket" + "github.com/astaxie/beego/session" + "github.com/feedlabs/elasticfeed/service/stream/model" ) +const ( + FEED_ADD = iota + FEED_DELETE + FEED_UPDATE + FEED_RESET + FEED_RELOAD + + ENTRY_ADD + ENTRY_DELETE + ENTRY_UPDATE + ENTRY_RELOAD +) + +var ( + Subscribe = make(chan Subscriber, 10) + Unsubscribe = make(chan string, 10) + Publish = make(chan model.Event, 10) + P2P = make(chan *websocket.Conn, 10) + + WaitingList = list.New() + Subscribers = list.New() + + GlobalSessions *session.Manager +) + type Subscription struct { Archive []model.Event New <-chan model.Event } +type Subscriber struct { + Name string + Conn *websocket.Conn +} + func NewEvent(ep model.EventType, user, msg string) model.Event { - return model.Event{ep, user, int(time.Now().Unix()), msg} + return model.Event{ep, user, time.Now().UnixNano(), msg} } -func Join(user string, ws *websocket.Conn) { - Subscribe <- Subscriber{Name: user, Conn: ws} +func NewChannelEvent(ep model.EventType, user, msg string) model.Event { + return NewEvent(ep, user, msg) } -func Leave(user string) { - Unsubscribe <- user +func NewSystemEvent(ep model.EventType, user, msg string) model.Event { + return NewChannelEvent(ep, user, msg) } -type Subscriber struct { - Name string - Conn *websocket.Conn +func NewFeedEvent(ep model.EventType, user, msg string) model.Event { + return NewSystemEvent(ep, user, msg) } -var ( - Subscribe = make(chan Subscriber, 10) - Unsubscribe = make(chan string, 10) - Publish = make(chan model.Event, 10) +func NewEntryEvent(ep model.EventType, user, msg string) model.Event { + return NewSystemEvent(ep, user, msg) +} - System_rpc = make(chan *websocket.Conn, 10) +func Join(user string, ws *websocket.Conn) { + Subscribe <- Subscriber{Name: user, Conn: ws} +} - WaitingList = list.New() - Subscribers = list.New() -) +func Leave(user string) { + Unsubscribe <- user +} func FeedManager() { for { @@ -50,10 +82,10 @@ func FeedManager() { case sub := <-Subscribe: Subscribers.PushBack(sub) - Publish <- NewEvent(model.EVENT_JOIN, sub.Name, "") + Publish <- NewChannelEvent(model.EVENT_JOIN, sub.Name, "") - case client := <-System_rpc: - data, _ := json.Marshal(&model.Event{model.EVENT_MESSAGE, "system", int(time.Now().Unix()), "ok"}) + case client := <-P2P: + data, _ := json.Marshal(NewSystemEvent(model.EVENT_MESSAGE, "system", "ok")) client.WriteMessage(websocket.TextMessage, data) case event := <-Publish: @@ -76,7 +108,7 @@ func FeedManager() { ws.Close() feedify.Error("WebSocket closed:", unsub) } - Publish <- NewEvent(model.EVENT_LEAVE, unsub, "") + Publish <- NewChannelEvent(model.EVENT_LEAVE, unsub, "") break } } @@ -84,7 +116,6 @@ func FeedManager() { } } - func broadcastWebSocket(event model.Event) { data, err := json.Marshal(event) if err != nil { @@ -103,5 +134,8 @@ func broadcastWebSocket(event model.Event) { } func init() { + GlobalSessions, _ = session.NewManager("memory", `{"cookieName":"elasticfeedsessid","gclifetime":3600}`) + go FeedManager() + go GlobalSessions.GC() } diff --git a/service/stream/model/event.go b/service/stream/model/event.go index 9abac77..eb8049b 100644 --- a/service/stream/model/event.go +++ b/service/stream/model/event.go @@ -13,28 +13,28 @@ const ( ) type Event struct { - Type EventType - User string - Timestamp int - Content string + Type EventType + User string + Timestamp int64 + Content string } -const archiveSize = 1 +const archiveSize = 100 -var archive = list.New() +var Archive = list.New() func NewArchive(event Event) { - if archive.Len() >= archiveSize { - archive.Remove(archive.Front()) + if Archive.Len() >= archiveSize { + Archive.Remove(Archive.Front()) } - archive.PushBack(event) + Archive.PushBack(event) } func GetEvents(lastReceived int) []Event { - events := make([]Event, 0, archive.Len()) - for event := archive.Front(); event != nil; event = event.Next() { + events := make([]Event, 0, Archive.Len()) + for event := Archive.Front(); event != nil; event = event.Next() { e := event.Value.(Event) - if e.Timestamp > int(lastReceived) { + if e.Timestamp > int64(lastReceived) { events = append(events, e) } } diff --git a/service/stream/router/default.go b/service/stream/router/default.go index acdcac1..acf4c10 100644 --- a/service/stream/router/default.go +++ b/service/stream/router/default.go @@ -6,11 +6,11 @@ import ( ) func init() { - feedify.SetStaticPath("/static", "stream/static") + feedify.SetStaticPath("/static", "service/stream/static") - feedify.Router("/lp/join", &channel.LongPollingController{}, "get:Join") - feedify.Router("/lp/post", &channel.LongPollingController{}) - feedify.Router("/lp/fetch", &channel.LongPollingController{}, "get:Fetch") + feedify.Router("/stream/lp/join", &channel.LongPollingController{}, "get:Join") + feedify.Router("/stream/lp/post", &channel.LongPollingController{}) + feedify.Router("/stream/lp/fetch", &channel.LongPollingController{}, "get:Fetch") - feedify.Router("/ws/join", &channel.WebSocketController{}, "get:Join") + feedify.Router("/stream/ws/join", &channel.WebSocketController{}, "get:Join") } diff --git a/service/stream/static/elasticfeed.js b/service/stream/static/elasticfeed.js index c910566..d592b96 100644 --- a/service/stream/static/elasticfeed.js +++ b/service/stream/static/elasticfeed.js @@ -1,7 +1,3 @@ -/* - * Author: Feed Labs - */ - function includeJs(jsFilePath) { var js = document.createElement("script"); @@ -14,70 +10,95 @@ function includeJs(jsFilePath) { includeJs('lib/feed.js'); includeJs('lib/entry.js'); includeJs('lib/channel.js'); -includeJs('lib/event.js'); +includeJs('lib/event/channel.js'); +includeJs('lib/event/system.js'); (function(window) { var elasticfeed = { /** @type {Object} */ - channel: null, + channelList: {}, - getFeed: function(id) { - return Feed; - }, + /** @type {Object} */ + feedList: {}, + + initFeed: function(options) { + options = _extend({ + channel: { + url: 'ws://localhost:80/ws', + transport: 'ws' + }, + styler: function(data) { + } + }, options); - getChannel: function() { - if (this.channel == null) { - this.channel = new Channel() - } + channel = this.getChannel(options.channel); - return this.channel; + return new Feed(options, channel); }, - newFeed: function(options) { - - }, + /** + * Returns Feed object + * @param options + * @returns {*} + */ + getFeed: function(options) { + if (options.id == undefined) { + return null; + } - newChannel: function(options) { + if (this.feedList[id] == undefined) { + this.feedList[id] = new Feed(options) + } + return this.feedList[options.id]; }, - load: function(url, callback) { - var xhr; - if (typeof XMLHttpRequest !== 'undefined') { - xhr = new XMLHttpRequest(); - } else { - var versions = ["MSXML2.XmlHttp.5.0", "MSXML2.XmlHttp.4.0", "MSXML2.XmlHttp.3.0", "MSXML2.XmlHttp.2.0", "Microsoft.XmlHttp"]; - for (var i = 0, len = versions.length; i < len; i++) { - try { - xhr = new ActiveXObject(versions[i]); - break; - } - catch (e) { - } - } + /** + * Returns Channel defined per API url + * @param options + * @param credential + * @returns {*} + */ + getChannel: function(options, credential) { + if (options.url == undefined) { + return null; } - xhr.onreadystatechange = ensureReadiness; - function ensureReadiness() { - if (xhr.readyState < 4) { - return; - } - if (xhr.status !== 200) { - return; - } - if (xhr.readyState === 4) { - callback(xhr); - } + + if (this.channelList[options.url] == undefined) { + this.channelList[options.url] = new Channel(options, credential) } - xhr.open('GET', url, true); - xhr.send(''); + return this.channelList[options.url]; + }, + + findFeed: function(id) { + return this.getFeed({id: id}); + }, + + findChannel: function(url) { + return this.getChannel({url: url}); } - }; // elasticfeed + }; - // =========================================================================== + // Helpers + + var _extend = function(a, b) { + var c = {}, prop; + for (prop in a) { + if (a.hasOwnProperty(prop)) { + c[prop] = a[prop]; + } + } + for (prop in b) { + if (b.hasOwnProperty(prop)) { + c[prop] = b[prop]; + } + } + return c; + } if ("function" === typeof define) { define(function(require) { @@ -86,15 +107,15 @@ includeJs('lib/event.js'); } else { window.elasticfeed = elasticfeed; } -}(window)); +}(window)); // Helpers -// ======= Element.prototype.remove = function() { this.parentElement.removeChild(this); }; + NodeList.prototype.remove = HTMLCollection.prototype.remove = function() { for (var i = 0, len = this.length; i < len; i++) { if (this[i] && this[i].parentElement) { diff --git a/service/stream/static/lib/channel.js b/service/stream/static/lib/channel.js index ce68ecd..8a6e711 100644 --- a/service/stream/static/lib/channel.js +++ b/service/stream/static/lib/channel.js @@ -1,13 +1,8 @@ var Channel = (function() { - const ACTION_JOIN = 0 - const ACTION_LEAVE = 1 - const ACTION_MESSAGE = 2 - - const AUTHENTICATED = 3 - const AUTHENTICATION_REQUIRED = 4 - const AUTHENTICATION_FAILED = 5 - const LOGGED_OUT = 6 + const JOIN = 0 + const LEAVE = 1 + const MESSAGE = 2 var defaultOptions = { id: null, @@ -15,17 +10,14 @@ var Channel = (function() { connectOnInit: true } - var defaultCredential = { - username: null, - token: null, - method: 'basic' - } - - function Channel(options, credential) { + function Channel(options) { /** @type {String} */ this.id = _uniqueId(); + /** @type {String} */ + this.url = null + /** @type {Object} */ this.options = _extend(defaultOptions, options); @@ -33,50 +25,90 @@ var Channel = (function() { this.id = this.options.id; } - /** @type {Object} */ - this.credential = _extend(defaultCredential, credential); + if (this.options.url != null) { + this.url = this.options.url; + } /** @type {Object} */ - this._handlers = []; + this._handlers = {}; } + // Handlers + /** - * @param {StreamEvent} event + * @param {ChannelEvent} event * @param {Function} callback */ - Channel.prototype.registerHandler = function(options, callback) { - options = { - id: null, - action_group: "feed|entry", - action_type: "add|del|update|*" + Channel.prototype.on = function(name, callback) { + switch (name) { + case 'join': + type = JOIN + break; + case 'leave': + type = LEAVE + break; + case 'message': + type = MESSAGE + break; + default: + return false; + break; } - - this._handlers.push({options: options, cb: callback}); + if (this._handlers[type] == undefined) { + this._handlers[type] = [] + } + this._handlers[type].push(callback); + return true; } + // Events + /** - * @param {StreamEvent} event + * @param {ChannelEvent} event */ Channel.prototype.onData = function(event) { - for (var i in this._handlers) { - this._handlers[i].cb.call(this, data); + switch (event.Type) { + case JOIN: + this.onJoin(event.User, event.ts) + break; + case LEAVE: + this.onLeave(event.User, event.ts) + break; + case MESSAGE: + this.onMessage(event.User, event.ts, event.Content) + break; } + } - this.EventToString(event); + Channel.prototype.onJoin = function(chid, timestamp) { + for (var i in this._handlers[JOIN]) { + this._handlers[JOIN][i].call(this, chid, timestamp); + } } - Channel.prototype.Authenticate = function(credential) { + Channel.prototype.onLeave = function(chid, timestamp) { + for (var i in this._handlers[LEAVE]) { + this._handlers[LEAVE][i].call(this, chid, timestamp); + } + } + + Channel.prototype.onMessage = function(chid, timestamp, data) { + for (var i in this._handlers[MESSAGE]) { + this._handlers[MESSAGE][i].call(this, chid, timestamp, data); + } } + // Connection + Channel.prototype.getConnection = function() { } Channel.prototype.getWebSocketConnection = function() { - this._socket = new WebSocket('ws://localhost:10100/ws/join?uname=' + this.id); + this._socket = new WebSocket('ws://localhost:10100/stream/ws/join?chid=' + this.id); self = this this._socket.onmessage = function(event) { - event = new StreamEvent(JSON.parse(event.data)) + event = new ChannelEvent(JSON.parse(event.data)) self.onData(event) }; @@ -92,7 +124,7 @@ var Channel = (function() { var lastReceived = 0; var isWait = false; - this.getJSON('http://localhost:10100/lp/join?uname=' + this.id, function() { + this.getJSON('http://localhost:10100/stream/lp/join?chid=' + this.id, function() { }) self = this; @@ -101,7 +133,7 @@ var Channel = (function() { return; } isWait = true; - self.getJSON("http://localhost:10100/lp/fetch?lastReceived=" + lastReceived, function(data, code) { + self.getJSON("http://localhost:10100/stream/lp/fetch?lastReceived=" + lastReceived, function(data, code) { if (code == 4) { isWait = false @@ -112,7 +144,7 @@ var Channel = (function() { } self.each(data, function(i, event) { - event = new StreamEvent(event) + event = new ChannelEvent(event) self.onData(event) lastReceived = event.GetTimestamp(); @@ -126,12 +158,44 @@ var Channel = (function() { return { send: function(data) { - self.post("/lp/post", {uname: self.id, content: JSON.stringify(data)}, function(status) { + self.post("/stream/lp/post", {chid: self.id, data: JSON.stringify(data)}, function(status) { }); } }; } + Channel.prototype.load = function(url, callback) { + var xhr; + if (typeof XMLHttpRequest !== 'undefined') { + xhr = new XMLHttpRequest(); + } else { + var versions = ["MSXML2.XmlHttp.5.0", "MSXML2.XmlHttp.4.0", "MSXML2.XmlHttp.3.0", "MSXML2.XmlHttp.2.0", "Microsoft.XmlHttp"]; + for (var i = 0, len = versions.length; i < len; i++) { + try { + xhr = new ActiveXObject(versions[i]); + break; + } + catch (e) { + } + } + } + xhr.onreadystatechange = ensureReadiness; + function ensureReadiness() { + if (xhr.readyState < 4) { + return; + } + if (xhr.status !== 200) { + return; + } + if (xhr.readyState === 4) { + callback(xhr); + } + } + + xhr.open('GET', url, true); + xhr.send(''); + } + // HTTP Channel.prototype.getJSON = function(url, callback) { @@ -182,13 +246,13 @@ var Channel = (function() { Channel.prototype.EventToString = function(event) { switch (event.Type) { - case ACTION_JOIN: + case JOIN: console.log(event.User + " joined the chat room"); break; - case ACTION_LEAVE: + case LEAVE: console.log(event.User + " left the chat room"); break; - case ACTION_MESSAGE: + case MESSAGE: console.log(event.User + ", " + event.PrintContent()); break; } diff --git a/service/stream/static/lib/event/channel.js b/service/stream/static/lib/event/channel.js new file mode 100644 index 0000000..dfbf253 --- /dev/null +++ b/service/stream/static/lib/event/channel.js @@ -0,0 +1,48 @@ +var ChannelEvent = (function() { + + function ChannelEvent(event) { + + /** @type {String} */ + this.id = null; + + /** @type {Integer} */ + this.ts = event.Timestamp; + + /** @type {Integer} */ + this.actionGroup = null + + /** @type {Integer} */ + this.actionType = null + + /** @type {String} */ + this.User = event.User + + /** @type {String} */ + this.Type = event.Type + + /** @type {String} */ + this.ContentType = 'string' + + /** @type {String} */ + try { + this.Content = JSON.parse(event.Content) + this.ContentType = 'json' + } catch (e) { + this.Content = event.Content + } + } + + ChannelEvent.prototype.GetTimestamp = function() { + return this.ts; + } + + ChannelEvent.prototype.PrintContent = function() { + if (this.ContentType == 'string') { + return this.Content + } + return JSON.stringify(this.Content) + } + + return ChannelEvent; + +})(); diff --git a/service/stream/static/lib/event.js b/service/stream/static/lib/event/entry.js similarity index 78% rename from service/stream/static/lib/event.js rename to service/stream/static/lib/event/entry.js index 34bba3a..fef5b97 100644 --- a/service/stream/static/lib/event.js +++ b/service/stream/static/lib/event/entry.js @@ -1,6 +1,6 @@ -var StreamEvent = (function() { +var EntryEvent = (function() { - function StreamEvent(event) { + function EntryEvent(event) { /** @type {String} */ this.id = null; @@ -32,17 +32,17 @@ var StreamEvent = (function() { } } - StreamEvent.prototype.GetTimestamp = function() { + EntryEvent.prototype.GetTimestamp = function() { return this.ts; } - StreamEvent.prototype.PrintContent = function() { + EntryEvent.prototype.PrintContent = function() { if (this.ContentType == 'string') { return this.Content } return JSON.stringify(this.Content) } - return StreamEvent; + return EntryEvent; })(); diff --git a/service/stream/static/lib/event/feed.js b/service/stream/static/lib/event/feed.js new file mode 100644 index 0000000..3b0432e --- /dev/null +++ b/service/stream/static/lib/event/feed.js @@ -0,0 +1,48 @@ +var FeedEvent = (function() { + + function FeedEvent(event) { + + /** @type {String} */ + this.id = null; + + /** @type {Integer} */ + this.ts = event.Timestamp; + + /** @type {Integer} */ + this.actionGroup = null + + /** @type {Integer} */ + this.actionType = null + + /** @type {String} */ + this.User = event.User + + /** @type {String} */ + this.Type = event.Type + + /** @type {String} */ + this.ContentType = 'string' + + /** @type {String} */ + try { + this.Content = JSON.parse(event.Content) + this.ContentType = 'json' + } catch (e) { + this.Content = event.Content + } + } + + FeedEvent.prototype.GetTimestamp = function() { + return this.ts; + } + + FeedEvent.prototype.PrintContent = function() { + if (this.ContentType == 'string') { + return this.Content + } + return JSON.stringify(this.Content) + } + + return FeedEvent; + +})(); diff --git a/service/stream/static/lib/event/system.js b/service/stream/static/lib/event/system.js new file mode 100644 index 0000000..2afedb6 --- /dev/null +++ b/service/stream/static/lib/event/system.js @@ -0,0 +1,48 @@ +var SystemEvent = (function() { + + function SystemEvent(event) { + + /** @type {String} */ + this.id = null; + + /** @type {Integer} */ + this.ts = event.Timestamp; + + /** @type {Integer} */ + this.actionGroup = null + + /** @type {Integer} */ + this.actionType = null + + /** @type {String} */ + this.User = event.User + + /** @type {String} */ + this.Type = event.Type + + /** @type {String} */ + this.ContentType = 'string' + + /** @type {String} */ + try { + this.Content = JSON.parse(event.Content) + this.ContentType = 'json' + } catch (e) { + this.Content = event.Content + } + } + + SystemEvent.prototype.GetTimestamp = function() { + return this.ts; + } + + SystemEvent.prototype.PrintContent = function() { + if (this.ContentType == 'string') { + return this.Content + } + return JSON.stringify(this.Content) + } + + return SystemEvent; + +})(); diff --git a/service/stream/static/lib/feed.js b/service/stream/static/lib/feed.js index 0d081b2..945dd1e 100644 --- a/service/stream/static/lib/feed.js +++ b/service/stream/static/lib/feed.js @@ -9,6 +9,11 @@ var Feed = (function() { const ACTION_HIDE = 5 const ACTION_SHOW = 6 + const AUTHENTICATED = 100 + const AUTHENTICATION_REQUIRED = 101 + const AUTHENTICATION_FAILED = 102 + const LOGGED_OUT = 103 + /** @type {Feed} */ var localCache = {} @@ -20,20 +25,40 @@ var Feed = (function() { defaultElementCount: 0 }; - function Feed(options, stylerFunction) { + /** @type {Object} */ + var globalCredential = { + username: null, + token: null, + method: 'basic' + }; + + function Feed(options, channel) { /** @type {String} */ this.id = null; - /** @type {String} */ - this.channelId = null; + /** @type {Channel} */ + this.channel = channel; /** @type {Array} */ this.entryList = []; + /** @type {Object} */ + if (this.channel.options.transport == 'ws') { + this.socket = this.channel.getWebSocketConnection(); + } else if (this.channel.options.transport == 'lp') { + this.socket = this.channel.getLongPoolingConnection(); + } + this.options = _extend(globalOptions, options); - this._stylerFunction = stylerFunction || this._stylerFunction; + this._stylerFunction = options.styler || this._stylerFunction; this.outputContainer = document.getElementById(this.options.outputContainerId); + + this.bindChannel(this.channel); + } + + Feed.prototype.on = function(type, callback) { + } // Events callbacks @@ -76,13 +101,12 @@ var Feed = (function() { // Handlers - Feed.prototype.registerHandlers = function() { - // bind to channel data - } - - // Channel management - - Feed.prototype.getChannel = function() { + Feed.prototype.bindChannel = function(channel) { + channel.on('message', function(chid, ts, data) { + // should detect type of message + // if feed addressed then check id + // trigger action if needed + }); } // Stylers diff --git a/service/stream/static/lp.html b/service/stream/static/lp.html index cc6b1ac..14a059e 100644 --- a/service/stream/static/lp.html +++ b/service/stream/static/lp.html @@ -4,7 +4,27 @@ diff --git a/service/stream/static/ws.html b/service/stream/static/ws.html index aa1c075..b658150 100644 --- a/service/stream/static/ws.html +++ b/service/stream/static/ws.html @@ -4,7 +4,24 @@ diff --git a/service/system/controller/status.go b/service/system/controller/status.go new file mode 100644 index 0000000..582e939 --- /dev/null +++ b/service/system/controller/status.go @@ -0,0 +1,45 @@ +package controller + +import ( + "os" + "runtime" + "strconv" + "github.com/feedlabs/feedify" + "github.com/feedlabs/elasticfeed/service/stream/controller/room" + "github.com/feedlabs/elasticfeed/service/stream/model" +) + +type StatusController struct { + feedify.Controller +} + +func (this *StatusController) Get() { + var memstats runtime.MemStats; + + runtime.ReadMemStats(&memstats) + hostname, _ := os.Hostname() + + this.Data["json"] = map[string]interface{}{ + "system": map[string]interface{} { + "hostname": hostname, + "pid": strconv.Itoa(os.Getpid()), + "cpus": strconv.Itoa(runtime.NumCPU()), + }, + "go": map[string]interface{} { + "routines": strconv.Itoa(runtime.NumGoroutine()), + }, + "mem": map[string]interface{} { + "mem_alloc": strconv.Itoa(int(memstats.Alloc)), + "mem_alloc_heap": strconv.Itoa(int(memstats.HeapAlloc)), + "mem_alloc_total": strconv.Itoa(int(memstats.TotalAlloc)), + "mem_sys": strconv.Itoa(int(memstats.Sys)), + }, + "stream": map[string]interface{} { + "subscribers": strconv.Itoa(room.Subscribers.Len()), + "waitinglist": strconv.Itoa(room.WaitingList.Len()), + "archived_queue": strconv.Itoa(model.Archive.Len()), + }, + } + + this.Controller.ServeJson() +} diff --git a/service/system/router/status.go b/service/system/router/status.go new file mode 100644 index 0000000..79a01a3 --- /dev/null +++ b/service/system/router/status.go @@ -0,0 +1,10 @@ +package router + +import ( + "github.com/feedlabs/feedify" + "github.com/feedlabs/elasticfeed/service/system/controller" +) + +func init() { + feedify.Router("/system/status", &controller.StatusController{}, "get:Get") +} diff --git a/service/system/system.go b/service/system/system.go new file mode 100644 index 0000000..f4352a0 --- /dev/null +++ b/service/system/system.go @@ -0,0 +1,8 @@ +package system + +import ( + _ "github.com/feedlabs/elasticfeed/service/system/controller" + _ "github.com/feedlabs/elasticfeed/service/system/router" +) + +func init() {}