go-music-player/music-player/pkg/server/clients.go

package server

import (
	"net/http"

	"github.com/felamaslen/go-music-player/pkg/logger"
	"github.com/go-redis/redis/v7"
)

func getClientNameFromRequest(r *http.Request) string {
  return r.URL.Query().Get(CLIENT_QUERY_NAME)
}

func endSubscription(sub *redis.PubSub) error {
  if err := sub.Unsubscribe(); err != nil {
    return err
  }
  if err := sub.Close(); err != nil {
    return err
  }
  return nil
}

func (c *Client) exposeToNetwork(l *logger.Logger, rdb *redis.Client) error {
  // Expose the client to all pods running the server
  if _, err := rdb.SAdd(KEY_CLIENT_NAMES, c.name).Result(); err != nil {
    return err
  }

  allClients, err := rdb.SMembers(KEY_CLIENT_NAMES).Result()
  if err != nil {
    return err
  }

  if err := publishAction(rdb, &Action{
    Type: ClientConnected,
    Payload: allClients,
  }); err != nil {
    return err
  }

  return nil
}

func (c *Client) disposeFromNetwork(l *logger.Logger, rdb *redis.Client) error {
  // Make sure other clients know when one goes away
  if _, err := rdb.SRem(KEY_CLIENT_NAMES, c.name).Result(); err != nil {
    return err
  }

  allClients, err := rdb.SMembers(KEY_CLIENT_NAMES).Result()
  if err != nil {
    return err
  }

  if err := publishAction(rdb, &Action{
    Type: ClientDisconnected,
    Payload: allClients,
  }); err != nil {
    return err
  }

  return nil
}

func (c *Client) subscribeToMe(l *logger.Logger, rdb *redis.Client) {
  // Subscribe this pod to messages from the client. This pod is responsible for
  // onward publishing to other pods where necessary, via internal pubsub

  for {
    select {
    case <- c.closeChan:
      return
    default:
      var actionFromClient Action
      if err := c.conn.ReadJSON(&actionFromClient); err != nil {
	return
      }

      l.Debug("[->Client] %s (%s)\n", actionFromClient.Type, c.name)

      actionFromClient.FromClient = &c.name

      if err := publishAction(rdb, &actionFromClient); err != nil {
	l.Error("Error publishing action from client: %v\n", err)
      }
    }
  }
}

func (c *Client) onConnect(l *logger.Logger, rdb *redis.Client) error {
  if err := c.exposeToNetwork(l, rdb); err != nil {
    l.Error("Error exposing new client to network: %v\n", err)
    return err
  }

  c.subscribeToMe(l, rdb)

  return nil
}

func (c *Client) onDisconnect(l *logger.Logger, rdb *redis.Client) error {
  l.Verbose("[Client disconnected] %s\n", c.name)
  close(c.closeChan)

  if err := c.disposeFromNetwork(l, rdb); err != nil {
    l.Error("Error disposing client from network: %v\n", err)
    return err
  }

  return nil
}