<commit_before>package websocket

import (
	"github.com/smancke/guble/protocol"
	"github.com/smancke/guble/server"
	"github.com/smancke/guble/store"

	"errors"
	"fmt"
	"math"
	"strconv"
	"strings"

	log "github.com/Sirupsen/logrus"
)

var errUnreadMsgsAvailable = errors.New("unread messages available")

// Receiver is a helper class, for managing a combined pull push on a topic.
// It is used for implementation of the + (receive) command in the guble protocol.
type Receiver struct {
	cancelC             chan bool
	sendC               chan []byte
	applicationId       string
	router              server.Router
	messageStore        store.MessageStore
	path                protocol.Path
	doFetch             bool
	doSubscription      bool
	startId             int64
	maxCount            int
	lastSendId          uint64
	shouldStop          bool
	route               *server.Route
	enableNotifications bool
	userId              string
}

// NewReceiverFromCmd parses the info in the command
func NewReceiverFromCmd(
	applicationId string,
	cmd *protocol.Cmd,
	sendChannel chan []byte,
	router server.Router,
	userId string) (rec *Receiver, err error) {

	messageStore, err := router.MessageStore()
	if err != nil {
		return nil, err
	}

	rec = &Receiver{
		applicationId:       applicationId,
		sendC:               sendChannel,
		router:              router,
		messageStore:        messageStore,
		cancelC:             make(chan bool, 1),
		enableNotifications: true,
		userId:              userId,
	}
	if len(cmd.Arg) == 0 || cmd.Arg[0] != '/' {
		return nil, fmt.Errorf("command requires at least a path argument, but non given")
	}

	args := strings.SplitN(cmd.Arg, " ", 3)
	rec.path = protocol.Path(args[0])

	if len(args) > 1 {
		rec.doFetch = true
		rec.startId, err = strconv.ParseInt(args[1], 10, 64)
		if err != nil {
			return nil, fmt.Errorf("startid has to be empty or int, but was %q: %v", args[1], err)
		}
	}

	rec.doSubscription = true
	if len(args) > 2 {
		rec.doSubscription = false
		rec.maxCount, err = strconv.Atoi(args[2])
		if err != nil {
			return nil, fmt.Errorf("maxCount has to be empty or int, but was %q: %v", args[1], err)
		}
	}

	return rec, nil
}

// Start starts the receiver loop
func (rec *Receiver) Start() error {
	rec.shouldStop = false
	if rec.doFetch && !rec.doSubscription {
		go rec.fetchOnlyLoop()
	} else {
		go rec.subscriptionLoop()
	}
	return nil
}

func (rec *Receiver) subscriptionLoop() {
	for !rec.shouldStop {
		if rec.doFetch {

			if err := rec.fetch(); err != nil {
				logger.WithError(err).WithField("rec", rec).Error("Error while fetching subscription")
				rec.sendError(protocol.ERROR_INTERNAL_SERVER, err.Error())
				return
			}

			if err := rec.messageStore.DoInTx(rec.path.Partition(), rec.subscribeIfNoUnreadMessagesAvailable); err != nil {
				if err == errUnreadMsgsAvailable {
					//fmt.Printf(" errUnreadMsgsAvailable lastSendId=%v\n", rec.lastSendId)
					rec.startId = int64(rec.lastSendId + 1)
					continue // fetch again
				} else {
					logger.WithError(err).WithField("recStartId", rec.startId).
						Error("Error while subscribeIfNoUnreadMessagesAvailable")
					rec.sendError(protocol.ERROR_INTERNAL_SERVER, err.Error())
					return
				}
			}
		}

		if !rec.doFetch {
			rec.subscribe()
		}
		rec.receiveFromSubscription()

		if !rec.shouldStop {
			//fmt.Printf(" router closed .. on msg: %v\n", rec.lastSendId)
			// the router kicked us out, because we are too slow for realtime listening,
			// so we setup parameters for fetching and closing the gap. Than we can subscribe again.
			rec.startId = int64(rec.lastSendId + 1)
			rec.doFetch = true
		}
	}
}

func (rec *Receiver) subscribeIfNoUnreadMessagesAvailable(maxMessageId uint64) error {
	if maxMessageId > rec.lastSendId {
		return errUnreadMsgsAvailable
	}
	rec.subscribe()
	return nil
}

func (rec *Receiver) subscribe() {
	rec.route = server.NewRoute(string(rec.path), rec.applicationId, rec.userId, 3)
	_, err := rec.router.Subscribe(rec.route)
	if err != nil {
		rec.sendError(protocol.ERROR_SUBSCRIBED_TO, string(rec.path), err.Error())
	} else {
		rec.sendOK(protocol.SUCCESS_SUBSCRIBED_TO, string(rec.path))
	}
}

func (rec *Receiver) receiveFromSubscription() {
	for {
		select {
		case m, ok := <-rec.route.MessagesChannel():
			if !ok {

				logger.WithFields(log.Fields{
					"applicationId": rec.applicationId,
				}).Debug("Router closed the channel returning from subscription for")
				return
			}

			logger.WithFields(log.Fields{<commit_msg>Fix receiver using NewRoute constructor<commit_after>package websocket

import (
	"github.com/smancke/guble/protocol"
	"github.com/smancke/guble/server"
	"github.com/smancke/guble/store"

	"errors"
	"fmt"
	"math"
	"strconv"
	"strings"

	log "github.com/Sirupsen/logrus"
)

var errUnreadMsgsAvailable = errors.New("unread messages available")

// Receiver is a helper class, for managing a combined pull push on a topic.
// It is used for implementation of the + (receive) command in the guble protocol.
type Receiver struct {
	cancelC             chan bool
	sendC               chan []byte
	applicationId       string
	router              server.Router
	messageStore        store.MessageStore
	path                protocol.Path
	doFetch             bool
	doSubscription      bool
	startId             int64
	maxCount            int
	lastSendId          uint64
	shouldStop          bool
	route               *server.Route
	enableNotifications bool
	userId              string
}

// NewReceiverFromCmd parses the info in the command
func NewReceiverFromCmd(
	applicationId string,
	cmd *protocol.Cmd,
	sendChannel chan []byte,
	router server.Router,
	userId string) (rec *Receiver, err error) {

	messageStore, err := router.MessageStore()
	if err != nil {
		return nil, err
	}

	rec = &Receiver{
		applicationId:       applicationId,
		sendC:               sendChannel,
		router:              router,
		messageStore:        messageStore,
		cancelC:             make(chan bool, 1),
		enableNotifications: true,
		userId:              userId,
	}
	if len(cmd.Arg) == 0 || cmd.Arg[0] != '/' {
		return nil, fmt.Errorf("command requires at least a path argument, but non given")
	}

	args := strings.SplitN(cmd.Arg, " ", 3)
	rec.path = protocol.Path(args[0])

	if len(args) > 1 {
		rec.doFetch = true
		rec.startId, err = strconv.ParseInt(args[1], 10, 64)
		if err != nil {
			return nil, fmt.Errorf("startid has to be empty or int, but was %q: %v", args[1], err)
		}
	}

	rec.doSubscription = true
	if len(args) > 2 {
		rec.doSubscription = false
		rec.maxCount, err = strconv.Atoi(args[2])
		if err != nil {
			return nil, fmt.Errorf("maxCount has to be empty or int, but was %q: %v", args[1], err)
		}
	}

	return rec, nil
}

// Start starts the receiver loop
func (rec *Receiver) Start() error {
	rec.shouldStop = false
	if rec.doFetch && !rec.doSubscription {
		go rec.fetchOnlyLoop()
	} else {
		go rec.subscriptionLoop()
	}
	return nil
}

func (rec *Receiver) subscriptionLoop() {
	for !rec.shouldStop {
		if rec.doFetch {

			if err := rec.fetch(); err != nil {
				logger.WithError(err).WithField("rec", rec).Error("Error while fetching subscription")
				rec.sendError(protocol.ERROR_INTERNAL_SERVER, err.Error())
				return
			}

			if err := rec.messageStore.DoInTx(rec.path.Partition(), rec.subscribeIfNoUnreadMessagesAvailable); err != nil {
				if err == errUnreadMsgsAvailable {
					//fmt.Printf(" errUnreadMsgsAvailable lastSendId=%v\n", rec.lastSendId)
					rec.startId = int64(rec.lastSendId + 1)
					continue // fetch again
				} else {
					logger.WithError(err).WithField("recStartId", rec.startId).
						Error("Error while subscribeIfNoUnreadMessagesAvailable")
					rec.sendError(protocol.ERROR_INTERNAL_SERVER, err.Error())
					return
				}
			}
		}

		if !rec.doFetch {
			rec.subscribe()
		}
		rec.receiveFromSubscription()

		if !rec.shouldStop {
			//fmt.Printf(" router closed .. on msg: %v\n", rec.lastSendId)
			// the router kicked us out, because we are too slow for realtime listening,
			// so we setup parameters for fetching and closing the gap. Than we can subscribe again.
			rec.startId = int64(rec.lastSendId + 1)
			rec.doFetch = true
		}
	}
}

func (rec *Receiver) subscribeIfNoUnreadMessagesAvailable(maxMessageId uint64) error {
	if maxMessageId > rec.lastSendId {
		return errUnreadMsgsAvailable
	}
	rec.subscribe()
	return nil
}

func (rec *Receiver) subscribe() {
	rec.route = server.NewRoute(
		server.RouteOptions{Path: rec.path, Size: 3},
		server.RouteParams{"application_id": rec.applicationId, "user_id": rec.userId},
	)
	_, err := rec.router.Subscribe(rec.route)
	if err != nil {
		rec.sendError(protocol.ERROR_SUBSCRIBED_TO, string(rec.path), err.Error())
	} else {
		rec.sendOK(protocol.SUCCESS_SUBSCRIBED_TO, string(rec.path))
	}
}

func (rec *Receiver) receiveFromSubscription() {
	for {
		select {
		case m, ok := <-rec.route.MessagesChannel():
			if !ok {

				logger.WithFields(log.Fields{
					"applicationId": rec.applicationId,
				}).Debug("Router closed the channel returning from subscription for")
				return
			}

			logger.WithFields(log.Fields{
{'fraction_non_alphanumeric': 0.11477746108309582, 'fraction_numerical': 0.006139004604253453, 'mean_word_length': 3.083706356311549, 'pattern_counts': {'":': 4, '<': 5, '<?xml version=': 0, '>': 9, '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': '898320', 'n_tokens_mistral': 3270, 'n_tokens_neox': 2957, 'n_words': 782}