<commit_before>package main

import (
	"crypto/rand"
	"encoding/base64"
	"fmt"
	"koding/broker/storage"
	"koding/tools/sockjs"
	"strconv"
	"strings"
	"time"

	"github.com/fatih/set"
	"github.com/streadway/amqp"
)

type Client struct {
	Session        *sockjs.Session
	ControlChannel *amqp.Channel
	SocketId       string
	Broker         *Broker
	LastPayload    string
	Subscriptions  storage.Subscriptionable
}

// NewClient retuns a new client that is defined on a given session.
func NewClient(session *sockjs.Session, broker *Broker) *Client {
	socketId := randomString()
	session.Tag = socketId

	var err error
	controlChannel, err := broker.PublishConn.Channel()
	if err != nil {
		panic(err)
	}

	var subscriptions storage.Subscriptionable

	subscriptions, err = storage.NewStorage(conf, STORAGE_BACKEND, socketId)
	if err != nil {
		log.Critical("Couldnt access to redis/create a key for client %v", session.Tag)
		STORAGE_BACKEND = "set"
		subscriptions, err = storage.NewStorage(conf, STORAGE_BACKEND, socketId)
		if err != nil {
			// this will never fail to here
			panic(err)
		}
	}

	globalMapMutex.Lock()
	sessionsMap[socketId] = session
	globalMapMutex.Unlock()

	return &Client{
		Session:        session,
		SocketId:       socketId,
		ControlChannel: controlChannel,
		Broker:         broker,
		Subscriptions:  subscriptions,
	}
}

// Close should be called whenever a client disconnects.
func (c *Client) Close() {
	log.Debug("Client Close Request for socketID: %v", c.SocketId)
	c.Subscriptions.Each(func(routingKeyPrefix interface{}) bool {
		c.RemoveFromRoute(routingKeyPrefix.(string))
		return true
	})

	c.Subscriptions.ClearWithTimeout()

	for {
		err := c.ControlChannel.Publish(c.Broker.AuthAllExchange, "broker.clientDisconnected", false, false, amqp.Publishing{Body: []byte(c.SocketId)})
		if err == nil {
			break
		}
		if amqpError, isAmqpError := err.(*amqp.Error); !isAmqpError || amqpError.Code != amqp.ChannelError {
			panic(err)
		}
		c.resetControlChannel()
	}

	log.Debug("Closing control channel for socketID: %v", c.SocketId)
	c.ControlChannel.Close()

	globalMapMutex.Lock()
	defer globalMapMutex.Unlock()
	delete(sessionsMap, c.SocketId)
}

// handleSessionMessage handles the received message from the client. It
// passes a response back to the client or publish the received message to a
// rabbitmq exchange for further process.
func (c *Client) handleSessionMessage(data interface{}) {
	defer log.RecoverAndLog()

	message := data.(map[string]interface{})
	log.Debug("Received message: %v", message)

	action := message["action"]
	switch action {
	case "subscribe":
		routingKeyPrefixes := strings.Split(message["routingKeyPrefix"].(string), " ")
		if err := c.Subscribe(routingKeyPrefixes...); err != nil {
			log.Error(err.Error())
		}

		sendToClient(c.Session, "broker.subscribed", message["routingKeyPrefix"])

	case "resubscribe":
		clientId := message["socketId"].(string)
		log.Debug("Resubscribe event for clientId: %v SocketId: %v", clientId, c.SocketId)
		found, err := c.Resubscribe(clientId)
		if err != nil {
			log.Error(err.Error())
		}
		log.Debug("Resubscribe found for socketID: %v, %v", clientId, found)
		sendToClient(c.Session, "broker.resubscribed", found)

	case "unsubscribe":
		routingKeyPrefixes := message["routingKeyPrefix"].(string)
		log.Debug("Unsubscribe event for socketID: %v, and prefixes", c.SocketId, routingKeyPrefixes)
		for _, routingKeyPrefix := range strings.Split(routingKeyPrefixes, " ") {
			c.Unsubscribe(routingKeyPrefix)
		}

	case "publish":
		exchange := message["exchange"].(string)
		routingKey := message["routingKey"].(string)
		payload := message["payload"].(string)

		log.Debug("Publish Event: Exchange: %v, RoutingKey %v, Payload %v",
			exchange,
			routingKey,
			payload,
		)

		err := c.Publish(exchange, routingKey, payload)
		if err != nil {
			log.Error(err.Error())
		}

	case "ping":
		sendToClient(c.Session, "broker.pong", nil)

	default:
		log.Warning("Invalid action. message: %v socketId: %v", message, c.SocketId)

	}
}

// Publish publish the given payload for to the given exchange and routingkey.
func (c *Client) Publish(exchange, routingKey, payload string) error {
	if !strings.HasPrefix(routingKey, "client.") {
		return fmt.Errorf("Invalid routing key: %v socketId: %v", routingKey, c.SocketId)
	}

	for {
		c.LastPayload = ""
		err := c.ControlChannel.Publish(exchange, routingKey, false, false, amqp.Publishing{CorrelationId: c.SocketId, Body: []byte(payload)})
		if err == nil {
			c.LastPayload = payload
			break
		}

		if amqpError, isAmqpError := err.(*amqp.Error); !isAmqpError || amqpError.Code != amqp.ChannelError {
			log.Warning("payload: %v routing key: %v exchange: %v err: %v",
				payload, routingKey, exchange, err)
		}

		time.Sleep(time.Second / 4) // penalty for crashing the AMQP channel
		c.resetControlChannel()
	}<commit_msg>Broker: expire subscription keys after 30 min<commit_after>package main

import (
	"crypto/rand"
	"encoding/base64"
	"fmt"
	"koding/broker/storage"
	"koding/tools/sockjs"
	"strconv"
	"strings"
	"time"

	"github.com/fatih/set"
	"github.com/streadway/amqp"
)

type Client struct {
	Session        *sockjs.Session
	ControlChannel *amqp.Channel
	SocketId       string
	Broker         *Broker
	LastPayload    string
	Subscriptions  storage.Subscriptionable
}

// NewClient retuns a new client that is defined on a given session.
func NewClient(session *sockjs.Session, broker *Broker) *Client {
	socketId := randomString()
	session.Tag = socketId

	var err error
	controlChannel, err := broker.PublishConn.Channel()
	if err != nil {
		panic(err)
	}

	var subscriptions storage.Subscriptionable

	subscriptions, err = storage.NewStorage(conf, STORAGE_BACKEND, socketId)
	if err != nil {
		log.Critical("Couldnt access to redis/create a key for client %v", session.Tag)
		STORAGE_BACKEND = "set"
		subscriptions, err = storage.NewStorage(conf, STORAGE_BACKEND, socketId)
		if err != nil {
			// this will never fail to here
			panic(err)
		}
	}

	globalMapMutex.Lock()
	sessionsMap[socketId] = session
	globalMapMutex.Unlock()

	return &Client{
		Session:        session,
		SocketId:       socketId,
		ControlChannel: controlChannel,
		Broker:         broker,
		Subscriptions:  subscriptions,
	}
}

// Close should be called whenever a client disconnects.
func (c *Client) Close() {
	log.Debug("Client Close Request for socketID: %v", c.SocketId)
	c.Subscriptions.Each(func(routingKeyPrefix interface{}) bool {
		c.RemoveFromRoute(routingKeyPrefix.(string))
		return true
	})

	c.Subscriptions.ClearWithTimeout(time.Minute * 5)

	for {
		err := c.ControlChannel.Publish(c.Broker.AuthAllExchange, "broker.clientDisconnected", false, false, amqp.Publishing{Body: []byte(c.SocketId)})
		if err == nil {
			break
		}
		if amqpError, isAmqpError := err.(*amqp.Error); !isAmqpError || amqpError.Code != amqp.ChannelError {
			panic(err)
		}
		c.resetControlChannel()
	}

	log.Debug("Closing control channel for socketID: %v", c.SocketId)
	c.ControlChannel.Close()

	globalMapMutex.Lock()
	defer globalMapMutex.Unlock()
	delete(sessionsMap, c.SocketId)
}

// handleSessionMessage handles the received message from the client. It
// passes a response back to the client or publish the received message to a
// rabbitmq exchange for further process.
func (c *Client) handleSessionMessage(data interface{}) {
	defer log.RecoverAndLog()

	message := data.(map[string]interface{})
	log.Debug("Received message: %v", message)

	action := message["action"]
	switch action {
	case "subscribe":
		routingKeyPrefixes := strings.Split(message["routingKeyPrefix"].(string), " ")
		if err := c.Subscribe(routingKeyPrefixes...); err != nil {
			log.Error(err.Error())
		}

		sendToClient(c.Session, "broker.subscribed", message["routingKeyPrefix"])

	case "resubscribe":
		clientId := message["socketId"].(string)
		log.Debug("Resubscribe event for clientId: %v SocketId: %v", clientId, c.SocketId)
		found, err := c.Resubscribe(clientId)
		if err != nil {
			log.Error(err.Error())
		}
		log.Debug("Resubscribe found for socketID: %v, %v", clientId, found)
		sendToClient(c.Session, "broker.resubscribed", found)

	case "unsubscribe":
		routingKeyPrefixes := message["routingKeyPrefix"].(string)
		log.Debug("Unsubscribe event for socketID: %v, and prefixes", c.SocketId, routingKeyPrefixes)
		for _, routingKeyPrefix := range strings.Split(routingKeyPrefixes, " ") {
			c.Unsubscribe(routingKeyPrefix)
		}

	case "publish":
		exchange := message["exchange"].(string)
		routingKey := message["routingKey"].(string)
		payload := message["payload"].(string)

		log.Debug("Publish Event: Exchange: %v, RoutingKey %v, Payload %v",
			exchange,
			routingKey,
			payload,
		)

		err := c.Publish(exchange, routingKey, payload)
		if err != nil {
			log.Error(err.Error())
		}

	case "ping":
		sendToClient(c.Session, "broker.pong", nil)
		// TOOD - may be we need to revisit this part later about duration and request count
		go c.Subscriptions.ClearWithTimeout(time.Minute * 30)
	default:
		log.Warning("Invalid action. message: %v socketId: %v", message, c.SocketId)

	}
}

// Publish publish the given payload for to the given exchange and routingkey.
func (c *Client) Publish(exchange, routingKey, payload string) error {
	if !strings.HasPrefix(routingKey, "client.") {
		return fmt.Errorf("Invalid routing key: %v socketId: %v", routingKey, c.SocketId)
	}

	for {
		c.LastPayload = ""
		err := c.ControlChannel.Publish(exchange, routingKey, false, false, amqp.Publishing{CorrelationId: c.SocketId, Body: []byte(payload)})
		if err == nil {
			c.LastPayload = payload
			break
		}

		if amqpError, isAmqpError := err.(*amqp.Error); !isAmqpError || amqpError.Code != amqp.ChannelError {
			log.Warning("payload: %v routing key: %v exchange: %v err: %v",
				payload, routingKey, exchange, err)
		}

		time.Sleep(time.Second / 4) // penalty for crashing the AMQP channel
		c.resetControlChannel()
	}
{'fraction_non_alphanumeric': 0.13008947421333064, 'fraction_numerical': 0.0011058610636372776, 'mean_word_length': 4.314102564102564, 'pattern_counts': {'":': 10, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4528960', 'n_tokens_mistral': 3513, 'n_tokens_neox': 3273, 'n_words': 825}