Another new feature: websocket controller, for real

Former-commit-id: c1a59b86733e890709b52446e22427a17d87f5fc
This commit is contained in:
Gerasimos (Makis) Maropoulos 2017-12-20 17:56:28 +02:00
parent b78698f6c0
commit 2042fddb66
10 changed files with 281 additions and 61 deletions

View File

@ -212,8 +212,10 @@ Follow the examples below,
- [Hello world](mvc/hello-world/main.go) **UPDATED** - [Hello world](mvc/hello-world/main.go) **UPDATED**
- [Session Controller](mvc/session-controller/main.go) **UPDATED** - [Session Controller](mvc/session-controller/main.go) **UPDATED**
- [Overview - Plus Repository and Service layers](mvc/overview) **NEW** - [Overview - Plus Repository and Service layers](mvc/overview) **UPDATED**
- [Login showcase - Plus Repository and Service layers](mvc/login) **NEW** - [Login showcase - Plus Repository and Service layers](mvc/login) **UPDATED**
- [Singleton](mvc/singleton) **NEW**
- [Websocket Controller](mvc/websocket) **NEW**
<!-- <!--
Why updated? Why updated?

View File

@ -0,0 +1,33 @@
package main
import (
"fmt"
"sync/atomic"
"github.com/kataras/iris"
"github.com/kataras/iris/mvc"
)
func main() {
app := iris.New()
mvc.New(app.Party("/")).Register(new(globalVisitorsController))
// http://localhost:8080
app.Run(iris.Addr(":8080"))
}
type globalVisitorsController struct {
// When a singleton controller is used then concurent safe access is up to the developers, because
// all clients share the same controller instance instead. Note that any controller's methods
// are per-client, but the struct's field can be shared accross multiple clients if the structure
// does not have any dynamic struct field depenendies that depend on the iris.Context,
// this declares a Singleton, note that you don't have to write a single line of code to do this, Iris is smart enough.
//
// see `Get`.
visits uint64
}
func (c *globalVisitorsController) Get() string {
count := atomic.AddUint64(&c.visits, 1)
return fmt.Sprintf("Total visitors: %d", count)
}

View File

@ -0,0 +1,82 @@
package main
import (
"sync/atomic"
"github.com/kataras/iris"
"github.com/kataras/iris/mvc"
"github.com/kataras/iris/websocket"
)
func main() {
app := iris.New()
// load templaes.
app.RegisterView(iris.HTML("./views", ".html"))
// render the ./views/index.html.
app.Get("/", func(ctx iris.Context) {
ctx.View("index.html")
})
mvc.Configure(app.Party("/websocket"), configureMVC)
// Or mvc.New(app.Party(...)).Configure(configureMVC)
// http://localhost:8080
app.Run(iris.Addr(":8080"))
}
func configureMVC(m *mvc.Application) {
ws := websocket.New(websocket.Config{})
// http://localhost:8080/websocket/iris-ws.js
m.Router.Any("/iris-ws.js", websocket.ClientHandler())
// This will bind the result of ws.Upgrade which is a websocket.Connection
// to the controller(s) registered via `m.Register`.
m.AddDependencies(ws.Upgrade)
m.Register(new(websocketController))
}
var visits uint64
func increment() uint64 {
return atomic.AddUint64(&visits, 1)
}
func decrement() uint64 {
return atomic.AddUint64(&visits, ^uint64(0))
}
type websocketController struct {
// Note that you could use an anonymous field as well, it doesn't matter, binder will find it.
//
// This is the current websocket connection, each client has its own instance of the *websocketController.
Conn websocket.Connection
}
func (c *websocketController) onLeave(roomName string) {
// visits--
newCount := decrement()
// This will call the "visit" event on all clients, except the current one,
// (it can't because it's left but for any case use this type of design)
c.Conn.To(websocket.Broadcast).Emit("visit", newCount)
}
func (c *websocketController) update() {
// visits++
newCount := increment()
// This will call the "visit" event on all clients, incuding the current
// with the 'newCount' variable.
//
// There are many ways that u can do it and faster, for example u can just send a new visitor
// and client can increment itself, but here we are just "showcasing" the websocket controller.
c.Conn.To(websocket.All).Emit("visit", newCount)
}
func (c *websocketController) Get( /* websocket.Connection could be lived here as well, it doesn't matter */ ) {
c.Conn.OnLeave(c.onLeave)
c.Conn.On("visit", c.update)
// call it after all event callbacks registration.
c.Conn.Wait()
}

View File

@ -0,0 +1,63 @@
<html>
<head>
<title>Online visitors MVC example</title>
<style>
body {
margin: 0;
font-family: -apple-system, "San Francisco", "Helvetica Neue", "Noto", "Roboto", "Calibri Light", sans-serif;
color: #212121;
font-size: 1.0em;
line-height: 1.6;
}
.container {
max-width: 750px;
margin: auto;
padding: 15px;
}
#online_visitors {
font-weight: bold;
font-size: 18px;
}
</style>
</head>
<body>
<div class="container">
<span id="online_visitors">1 online visitor</span>
</div>
<script src="/websocket/iris-ws.js"></script>
<script type="text/javascript">
(function () {
var socket = new Ws("ws://localhost:8080/websocket");
socket.OnConnect(function(){
// update the rest of connected clients, including "myself" when "my" connection is 100% ready.
socket.Emit("visit");
});
socket.On("visit", function (newCount) {
console.log("visit websocket event with newCount of: ", newCount);
var text = "1 online visitor";
if (newCount > 1) {
text = newCount + " online visitors";
}
document.getElementById("online_visitors").innerHTML = text;
});
socket.OnDisconnect(function () {
document.getElementById("online_visitors").innerHTML = "you've been disconnected";
});
})();
</script>
</body>
</html>

View File

@ -47,8 +47,9 @@ func handleConnection(c websocket.Connection) {
c.On("chat", func(msg string) { c.On("chat", func(msg string) {
// Print the message to the console, c.Context() is the iris's http context. // Print the message to the console, c.Context() is the iris's http context.
fmt.Printf("%s sent: %s\n", c.Context().RemoteAddr(), msg) fmt.Printf("%s sent: %s\n", c.Context().RemoteAddr(), msg)
// Write message back to the client message owner: // Write message back to the client message owner with:
// c.Emit("chat", msg) // c.Emit("chat", msg)
// Write message to all except this client with:
c.To(websocket.Broadcast).Emit("chat", msg) c.To(websocket.Broadcast).Emit("chat", msg)
}) })
} }

View File

@ -105,8 +105,9 @@ var Ws = (function () {
t = websocketJSONMessageType; t = websocketJSONMessageType;
m = JSON.stringify(data); m = JSON.stringify(data);
} }
else { else if (data !== null && typeof(data) !== "undefined" ) {
console.log("Invalid, javascript-side should contains an empty second parameter."); // if it has a second parameter but it's not a type we know, then fire this:
console.log("unsupported type of input argument passed, try to not include this argument to the 'Emit'");
} }
return this._msg(event, t, m); return this._msg(event, t, m);
}; };

View File

@ -106,8 +106,9 @@ class Ws {
//propably json-object //propably json-object
t = websocketJSONMessageType; t = websocketJSONMessageType;
m = JSON.stringify(data); m = JSON.stringify(data);
} else { } else if (data !== null && typeof (data) !== "undefined") {
console.log("Invalid, javascript-side should contains an empty second parameter."); // if it has a second parameter but it's not a type we know, then fire this:
console.log("unsupported type of input argument passed, try to not include this argument to the 'Emit'");
} }
return this._msg(event, t, m); return this._msg(event, t, m);

View File

@ -137,6 +137,9 @@ type (
Connection interface { Connection interface {
// Emitter implements EmitMessage & Emit // Emitter implements EmitMessage & Emit
Emitter Emitter
// Err is not nil if the upgrader failed to upgrade http to websocket connection.
Err() error
// ID returns the connection's identifier // ID returns the connection's identifier
ID() string ID() string
@ -181,6 +184,11 @@ type (
// Note: the callback(s) called right before the server deletes the connection from the room // Note: the callback(s) called right before the server deletes the connection from the room
// so the connection theoretical can still send messages to its room right before it is being disconnected. // so the connection theoretical can still send messages to its room right before it is being disconnected.
OnLeave(roomLeaveCb LeaveRoomFunc) OnLeave(roomLeaveCb LeaveRoomFunc)
// Wait starts the pinger and the messages reader,
// it's named as "Wait" because it should be called LAST,
// after the "On" events IF server's `Upgrade` is used,
// otherise you don't have to call it because the `Handler()` does it automatically.
Wait()
// Disconnect disconnects the client, close the underline websocket conn and removes it from the conn list // Disconnect disconnects the client, close the underline websocket conn and removes it from the conn list
// returns the error, if any, from the underline connection // returns the error, if any, from the underline connection
Disconnect() error Disconnect() error
@ -197,6 +205,7 @@ type (
} }
connection struct { connection struct {
err error
underline UnderlineConnection underline UnderlineConnection
id string id string
messageType int messageType int
@ -207,6 +216,7 @@ type (
onPingListeners []PingFunc onPingListeners []PingFunc
onNativeMessageListeners []NativeMessageFunc onNativeMessageListeners []NativeMessageFunc
onEventListeners map[string][]MessageFunc onEventListeners map[string][]MessageFunc
started bool
// these were maden for performance only // these were maden for performance only
self Emitter // pre-defined emitter than sends message to its self client self Emitter // pre-defined emitter than sends message to its self client
broadcast Emitter // pre-defined emitter that sends message to all except this broadcast Emitter // pre-defined emitter that sends message to all except this
@ -237,6 +247,7 @@ func newConnection(ctx context.Context, s *Server, underlineConn UnderlineConnec
onErrorListeners: make([]ErrorFunc, 0), onErrorListeners: make([]ErrorFunc, 0),
onNativeMessageListeners: make([]NativeMessageFunc, 0), onNativeMessageListeners: make([]NativeMessageFunc, 0),
onEventListeners: make(map[string][]MessageFunc, 0), onEventListeners: make(map[string][]MessageFunc, 0),
started: false,
ctx: ctx, ctx: ctx,
server: s, server: s,
} }
@ -252,6 +263,11 @@ func newConnection(ctx context.Context, s *Server, underlineConn UnderlineConnec
return c return c
} }
// Err is not nil if the upgrader failed to upgrade http to websocket connection.
func (c *connection) Err() error {
return c.err
}
// write writes a raw websocket message with a specific type to the client // write writes a raw websocket message with a specific type to the client
// used by ping messages and any CloseMessage types. // used by ping messages and any CloseMessage types.
func (c *connection) write(websocketMessageType int, data []byte) error { func (c *connection) write(websocketMessageType int, data []byte) error {
@ -322,6 +338,13 @@ func (c *connection) startPinger() {
}() }()
} }
func (c *connection) fireOnPing() {
// fire the onPingListeners
for i := range c.onPingListeners {
c.onPingListeners[i]()
}
}
func (c *connection) startReader() { func (c *connection) startReader() {
conn := c.underline conn := c.underline
hasReadTimeout := c.server.config.ReadTimeout > 0 hasReadTimeout := c.server.config.ReadTimeout > 0
@ -503,11 +526,20 @@ func (c *connection) fireOnLeave(roomName string) {
} }
} }
func (c *connection) fireOnPing() { // Wait starts the pinger and the messages reader,
// fire the onPingListeners // it's named as "Wait" because it should be called LAST,
for i := range c.onPingListeners { // after the "On" events IF server's `Upgrade` is used,
c.onPingListeners[i]() // otherise you don't have to call it because the `Handler()` does it automatically.
func (c *connection) Wait() {
if c.started {
return
} }
c.started = true
// start the ping
c.startPinger()
// start the messages reader
c.startReader()
} }
func (c *connection) Disconnect() error { func (c *connection) Disconnect() error {

View File

@ -1,10 +1,10 @@
package websocket package websocket
const ( const (
// All is the string which the Emitter use to send a message to all // All is the string which the Emitter use to send a message to all.
All = "" All = ""
// Broadcast is the string which the Emitter use to send a message to all except this connection // Broadcast is the string which the Emitter use to send a message to all except this connection.
Broadcast = ";ionwebsocket;to;all;except;me;" Broadcast = ";to;all;except;me;"
) )
type ( type (

View File

@ -109,7 +109,7 @@ type (
mu sync.RWMutex // for rooms mu sync.RWMutex // for rooms
onConnectionListeners []ConnectionFunc onConnectionListeners []ConnectionFunc
//connectionPool sync.Pool // sadly we can't make this because the websocket connection is live until is closed. //connectionPool sync.Pool // sadly we can't make this because the websocket connection is live until is closed.
handler context.Handler upgrader websocket.Upgrader
} }
) )
@ -119,10 +119,20 @@ type (
// //
// To serve the built'n javascript client-side library look the `websocket.ClientHandler`. // To serve the built'n javascript client-side library look the `websocket.ClientHandler`.
func New(cfg Config) *Server { func New(cfg Config) *Server {
cfg = cfg.Validate()
return &Server{ return &Server{
config: cfg.Validate(), config: cfg,
rooms: make(map[string][]string, 0), rooms: make(map[string][]string, 0),
onConnectionListeners: make([]ConnectionFunc, 0), onConnectionListeners: make([]ConnectionFunc, 0),
upgrader: websocket.Upgrader{
HandshakeTimeout: cfg.HandshakeTimeout,
ReadBufferSize: cfg.ReadBufferSize,
WriteBufferSize: cfg.WriteBufferSize,
Error: cfg.Error,
CheckOrigin: cfg.CheckOrigin,
Subprotocols: cfg.Subprotocols,
EnableCompression: cfg.EnableCompression,
},
} }
} }
@ -135,40 +145,50 @@ func New(cfg Config) *Server {
// //
// To serve the built'n javascript client-side library look the `websocket.ClientHandler`. // To serve the built'n javascript client-side library look the `websocket.ClientHandler`.
func (s *Server) Handler() context.Handler { func (s *Server) Handler() context.Handler {
// build the upgrader once
c := s.config
upgrader := websocket.Upgrader{
HandshakeTimeout: c.HandshakeTimeout,
ReadBufferSize: c.ReadBufferSize,
WriteBufferSize: c.WriteBufferSize,
Error: c.Error,
CheckOrigin: c.CheckOrigin,
Subprotocols: c.Subprotocols,
EnableCompression: c.EnableCompression,
}
return func(ctx context.Context) { return func(ctx context.Context) {
// Upgrade upgrades the HTTP Server connection to the WebSocket protocol. c := s.Upgrade(ctx)
// // NOTE TO ME: fire these first BEFORE startReader and startPinger
// The responseHeader is included in the response to the client's upgrade // in order to set the events and any messages to send
// request. Use the responseHeader to specify cookies (Set-Cookie) and the // the startPinger will send the OK to the client and only
// application negotiated subprotocol (Sec--Protocol). // then the client is able to send and receive from Server
// // when all things are ready and only then. DO NOT change this order.
// If the upgrade fails, then Upgrade replies to the client with an HTTP error
// response. // fire the on connection event callbacks, if any
conn, err := upgrader.Upgrade(ctx.ResponseWriter(), ctx.Request(), ctx.ResponseWriter().Header()) for i := range s.onConnectionListeners {
if err != nil { s.onConnectionListeners[i](c)
ctx.Application().Logger().Warnf("websocket error: %v\n", err)
ctx.StatusCode(503) // Status Service Unavailable
return
} }
s.handleConnection(ctx, conn)
// start the ping and the messages reader
c.Wait()
} }
} }
// handleConnection creates & starts to listening to a new connection // Upgrade upgrades the HTTP Server connection to the WebSocket protocol.
func (s *Server) handleConnection(ctx context.Context, websocketConn UnderlineConnection) { //
// The responseHeader is included in the response to the client's upgrade
// request. Use the responseHeader to specify cookies (Set-Cookie) and the
// application negotiated subprotocol (Sec--Protocol).
//
// If the upgrade fails, then Upgrade replies to the client with an HTTP error
// response and the return `Connection.Err()` is filled with that error.
//
// For a more high-level function use the `Handler()` and `OnConnecton` events.
// This one does not starts the connection's writer and reader, so after your `On/OnMessage` events registration
// the caller has to call the `Connection#Wait` function, otherwise the connection will be not handled.
func (s *Server) Upgrade(ctx context.Context) Connection {
conn, err := s.upgrader.Upgrade(ctx.ResponseWriter(), ctx.Request(), ctx.ResponseWriter().Header())
if err != nil {
ctx.Application().Logger().Warnf("websocket error: %v\n", err)
ctx.StatusCode(503) // Status Service Unavailable
return &connection{err: err}
}
return s.handleConnection(ctx, conn)
}
// wrapConnection wraps an underline connection to an iris websocket connection.
// It does NOT starts its writer, reader and event mux, the caller is responsible for that.
func (s *Server) handleConnection(ctx context.Context, websocketConn UnderlineConnection) *connection {
// use the config's id generator (or the default) to create a websocket client/connection id // use the config's id generator (or the default) to create a websocket client/connection id
cid := s.config.IDGenerator(ctx) cid := s.config.IDGenerator(ctx)
// create the new connection // create the new connection
@ -179,22 +199,7 @@ func (s *Server) handleConnection(ctx context.Context, websocketConn UnderlineCo
// join to itself // join to itself
s.Join(c.ID(), c.ID()) s.Join(c.ID(), c.ID())
// NOTE TO ME: fire these first BEFORE startReader and startPinger return c
// in order to set the events and any messages to send
// the startPinger will send the OK to the client and only
// then the client is able to send and receive from Server
// when all things are ready and only then. DO NOT change this order.
// fire the on connection event callbacks, if any
for i := range s.onConnectionListeners {
s.onConnectionListeners[i](c)
}
// start the ping
c.startPinger()
// start the messages reader
c.startReader()
} }
/* Notes: /* Notes: