func (c *ChatterBox) clientSender(ctx context.Context, usr string, chName string, conn *websocket.Conn, store *boutique.Store) {
	const (
		msgField   = "Messages"
		usersField = "Users"
	)
	state := store.State()
	startData := state.Data.(data.State)
	var lastMsgID = -1
	if len(startData.Messages) > 0 {
		lastMsgID = startData.Messages[len(startData.Messages)-1].ID
	}
	msgCh, msgCancel, err := store.Subscribe(msgField)
	if err != nil {
		c.sendError(conn, err)
		return
	}
	defer msgCancel()
	usersCh, usersCancel, err := store.Subscribe(usersField)
	if err != nil {
		c.sendError(conn, err)
		return
	}
	defer usersCancel()
	for {
		select {
		case msgSig := <-msgCh:
			msgs := msgSig.State.Data.(data.State).Messages
			if len(msgs) == 0 {  
				continue
			}
			var toSend []data.Message
			toSend, lastMsgID = c.latestMsgs(msgs, lastMsgID)
			if len(toSend) > 0 {
				if err := c.sendMessages(conn, toSend); err != nil {
					glog.Errorf("error sending message to client on channel %s: %s", chName, err)
					return
				}
			}
		case userSig := <-usersCh:
			if err := c.write(conn, messages.Server{Type: messages.SMUserUpdate, Users: userSig.State.Data.(data.State).Users}); err != nil {
				c.sendError(conn, err)
				return
			}
		case <-ctx.Done():
			return
		}
	}
}
clientSender receives changes to the store's Messages field and pushes them out to
our websocket clients.
package server
import (
	"context"
	"fmt"
	"net/http"
	"sync"
	"time"
	"github.com/golang/glog"
	"github.com/gorilla/websocket"
	"github.com/johnsiilver/boutique"
	"github.com/johnsiilver/boutique/example/chatterbox/messages"
	"github.com/johnsiilver/boutique/example/chatterbox/server/state"
	"github.com/johnsiilver/boutique/example/chatterbox/server/state/actions"
	"github.com/johnsiilver/boutique/example/chatterbox/server/state/data"
)
var upgrader = websocket.Upgrader{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
}
type channel struct {
	ctx    context.Context
	cancel context.CancelFunc
	hub    *state.Hub
	users  map[string]bool
}
type ChatterBox struct {
	chMu     sync.RWMutex
	channels map[string]*channel
}
func New() *ChatterBox {
	return &ChatterBox{channels: map[string]*channel{}}
}
func (c *ChatterBox) Handler(w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		glog.Errorf("error connecting to server: %s", err)
		return
	}
	wg := &sync.WaitGroup{}
	wg.Add(2)
	go c.clientReceiver(r.Context(), wg, conn)
	wg.Wait()
}
func (c *ChatterBox) subscribe(ctx context.Context, cancel context.CancelFunc, conn *websocket.Conn, m messages.Client) (*state.Hub, error) {
	c.chMu.Lock()
	defer c.chMu.Unlock()
	var (
		hub *state.Hub
		err error
	)
	mchan, ok := c.channels[m.Channel]
	if ok {
		hub = mchan.hub
		if mchan.users[m.User] {
			c.sendError(
				conn,
				fmt.Errorf("a user named %s is already in this channel: %s", m.User, m.Channel),
			)
			return nil, fmt.Errorf("subscribe erorr")
		}
	} else {
		hub, err = state.New(m.Channel)
		if err != nil {
			return nil, err
		}
		mchan = &channel{ctx: ctx, cancel: cancel, hub: hub, users: map[string]bool{m.User: true}}
		c.channels[m.Channel] = mchan
	}
	mchan.users[m.User] = true
	if err = hub.Store.Perform(actions.AddUser(m.User)); err != nil {
		return nil, err
	}
	err = c.write(
		conn,
		messages.Server{
			Type:  messages.SMSubAck,
			Users: hub.Store.State().Data.(data.State).Users,
		},
	)
	if err != nil {
		glog.Errorf("problem writing subAck for user %s to chan %s: %s", m.User, m.Channel, err)
		delete(mchan.users, m.User)
		if len(mchan.users) == 0 {
			delete(c.channels, m.Channel)
		}
		return nil, fmt.Errorf("could not subscribe user %s: %s", m.User, err)
	}
	glog.Infof("client %v: subscribed to %s as %s", conn.RemoteAddr(), m.User, m.Channel)
	return hub, nil
}
func (c *ChatterBox) unsubscribe(u string, channel string) {
	glog.Infof("called unsubsribe")
	c.chMu.Lock()
	defer c.chMu.Unlock()
	mchan, ok := c.channels[channel]
	if !ok {
		glog.Infof("could not find channel %s", channel)
		return
	}
	mchan.cancel()
	delete(mchan.users, u)
	if err := mchan.hub.Store.Perform(actions.RemoveUser(u)); err != nil {
		glog.Errorf("problem removing user from Store: %s", err)
	}
	glog.Infof("unsubscribed %s from %s", u, channel)
}
func (c *ChatterBox) clientReceiver(ctx context.Context, wg *sync.WaitGroup, conn *websocket.Conn) {
	defer wg.Done()
	var (
		cancel context.CancelFunc
		hub    *state.Hub
		user   string
		comm   string
	)
	for {
		m, err := c.read(conn)
		if err != nil {
			glog.Errorf("client %s terminated its connection", conn.RemoteAddr())
			if cancel != nil {
				cancel()
			}
			return
		}
		err = m.Validate()
		if err != nil {
			glog.Errorf("error: client %v message did not validate: %v: %#+v: ignoring...", conn.RemoteAddr(), err, m)
			if err = c.sendError(conn, err); err != nil {
				return
			}
			continue
		}
		switch t := m.Type; t {
		case messages.CMSendText:
			if hub == nil {
				if err := c.sendError(conn, fmt.Errorf("cannot send a message, not subscribed to channel")); err != nil {
					glog.Errorf("error: client %v: %s", conn.RemoteAddr(), err)
					return
				}
			}
			if err := hub.Store.Perform(actions.SendMessage(user, m.Text.Text)); err != nil {
				c.sendError(conn, fmt.Errorf("problem calling store.Perform(): %s", err))
				continue
			}
		case messages.CMSubscribe:
			ctx, cancel = context.WithCancel(context.Background())
			defer cancel()
			if hub != nil {
				c.unsubscribe(user, comm)
				if err := c.write(conn, messages.Server{Type: messages.SMChannelDrop}); err != nil {
					glog.Errorf("error: client %v: %s", conn.RemoteAddr(), err)
					return
				}
				if cancel != nil {
					cancel()
				}
			}
			var err error
			hub, err = c.subscribe(ctx, cancel, conn, m)
			if err != nil {
				if err = c.sendError(conn, err); err != nil {
					glog.Errorf("error: client %v: %s", conn.RemoteAddr(), err)
				}
				return
			}
			user = m.User
			comm = m.Channel
			defer c.unsubscribe(user, comm)
			go c.clientSender(ctx, user, comm, conn, hub.Store)
		case messages.CMDrop:
			if hub == nil {
				if err := c.sendError(conn, fmt.Errorf("error: cannot drop a channel, your not subscribed to any")); err != nil {
					glog.Errorf("error: client %v: %s", conn.RemoteAddr(), err)
					return
				}
			}
			if cancel != nil {
				cancel()
			}
			c.unsubscribe(user, comm)
			if err := c.write(conn, messages.Server{Type: messages.SMChannelDrop}); err != nil {
				glog.Errorf("error: client %v: %s", conn.RemoteAddr(), err)
				return
			}
			hub = nil
			user = ""
			comm = ""
		default:
			glog.Errorf("error: client %v had unknown message %v, ignoring", conn.RemoteAddr(), t)
			if err := c.sendError(conn, fmt.Errorf("received message type from client %v that the server doesn't understand", t)); err != nil {
				glog.Errorf("error: client %v: %s", conn.RemoteAddr(), err)
				return
			}
		}
	}
}