Realtime chat: Avoid concurrent access to the map (gin-gonic#22)

Realtime chat: Avoid concurrent access to the map

realtime-chat/main.go

@@ -5,11 +5,13 @@ import (
  "io"
  "math/rand"
  "net/http"
-
  "github.com/gin-gonic/gin"
 )
 
+var roomManager *Manager
+
 func main() {
+ roomManager = NewRoomManager()
  router := gin.Default()
  router.SetHTMLTemplate(html)
 
@@ -23,8 +25,8 @@ func main() {
 
 func stream(c *gin.Context) {
  roomid := c.Param("roomid")
- listener := openListener(roomid)
- defer closeListener(roomid, listener)
+ listener := roomManager.OpenListener(roomid)
+ defer roomManager.CloseListener(roomid, listener)
 
  clientGone := c.Writer.CloseNotify()
  c.Stream(func(w io.Writer) bool {
@@ -51,7 +53,7 @@ func roomPOST(c *gin.Context) {
  roomid := c.Param("roomid")
  userid := c.PostForm("user")
  message := c.PostForm("message")
- room(roomid).Submit(userid + ": " + message)
+ roomManager.Submit(userid, roomid, message)
 
  c.JSON(http.StatusOK, gin.H{
  "status": "success",
@@ -61,5 +63,5 @@ func roomPOST(c *gin.Context) {
 
 func roomDELETE(c *gin.Context) {
  roomid := c.Param("roomid")
- deleteBroadcast(roomid)
+ roomManager.DeleteBroadcast(roomid)
 }

realtime-chat/rooms.go

@@ -1,33 +1,107 @@
 package main
 
-import "github.com/dustin/go-broadcast"
+import (
+ "github.com/dustin/go-broadcast"
+)
 
-var roomChannels = make(map[string]broadcast.Broadcaster)
+type Message struct {
+ UserId string
+ RoomId string
+ Text string
+}
 
-func openListener(roomid string) chan interface{} {
- listener := make(chan interface{})
- room(roomid).Register(listener)
- return listener
+type Listener struct {
+ RoomId string
+ Chan chan interface{}
+}
+
+type Manager struct {
+ roomChannels map[string]broadcast.Broadcaster
+ open chan *Listener
+ close chan *Listener
+ delete chan string
+ messages chan *Message
+}
+
+func NewRoomManager() *Manager {
+ manager := &Manager{
+ roomChannels: make(map[string]broadcast.Broadcaster),
+ open: make(chan *Listener, 100),
+ close: make(chan *Listener, 100),
+ delete: make(chan string, 100),
+ messages: make(chan *Message, 100),
+ }
+
+ go manager.run()
+ return manager
+}
+
+func (m *Manager) run() {
+ for {
+ select {
+ case listener := <- m.open:
+ m.register(listener)
+ case listener := <- m.close:
+ m.deregister(listener)
+ case roomid := <- m.delete:
+ m.deleteBroadcast(roomid)
+ case message := <- m.messages:
+ m.room(message.RoomId).Submit(message.UserId + ": " + message.Text)
+ }
+ }
+}
+
+func (m *Manager) register(listener *Listener) {
+ m.room(listener.RoomId).Register(listener.Chan)
 }
 
-func closeListener(roomid string, listener chan interface{}) {
- room(roomid).Unregister(listener)
- close(listener)
+func (m *Manager) deregister(listener *Listener) {
+ m.room(listener.RoomId).Unregister(listener.Chan)
+ close(listener.Chan)
 }
 
-func deleteBroadcast(roomid string) {
- b, ok := roomChannels[roomid]
+func (m *Manager) deleteBroadcast(roomid string) {
+ b, ok := m.roomChannels[roomid]
  if ok {
  b.Close()
- delete(roomChannels, roomid)
+ delete(m.roomChannels, roomid)
  }
 }
 
-func room(roomid string) broadcast.Broadcaster {
- b, ok := roomChannels[roomid]
+func (m *Manager) room(roomid string) broadcast.Broadcaster {
+ b, ok := m.roomChannels[roomid]
  if !ok {
  b = broadcast.NewBroadcaster(10)
- roomChannels[roomid] = b
+ m.roomChannels[roomid] = b
  }
  return b
 }
+
+func (m *Manager) OpenListener(roomid string) chan interface{}{
+ listener := make(chan interface{})
+ m.open <- &Listener{
+ RoomId: roomid,
+ Chan: listener,
+ }
+ return listener
+}
+
+func (m *Manager) CloseListener(roomid string, channel chan interface{}) {
+ m.close <- &Listener{
+ RoomId: roomid,
+ Chan: channel,
+ }
+}
+
+func (m *Manager) DeleteBroadcast(roomid string) {
+ m.delete <- roomid
+}
+
+func (m *Manager) Submit (userid, roomid, text string) {
+ msg := &Message{
+ UserId: userid,
+ RoomId: roomid,
+ Text: text,
+ }
+ m.messages <- msg
+}