<commit_before>package drop

import (
	"container/ring"
	"encoding/base64"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"os"
	"time"
)

const (
	MaximumMessageSize = 4096
	dropIdLength       = 32
)

type MessageSizeExceededError struct{}

func (m MessageSizeExceededError) Error() string {
	return fmt.Sprintf("drop: maximum message size of %d bytes exceeded", MaximumMessageSize)
}

type Drop struct {
	Id string
}

func (d Drop) Verify() bool {
	// must be 32 byte encoded to 43 base64url characters
	bytes, err := base64.URLEncoding.DecodeString(d.Id + "=") // add padding to make parser happy
	return err == nil && len(bytes) == dropIdLength
}

type message struct {
	drop      Drop
	timestamp time.Time
	path      string
}

func (msg *message) discard() error {
	return os.Remove(msg.path)
}

type DropManager struct {
	capacity   int
	baseDir    string
	msgRing    *ring.Ring
	submitChan chan message
}

func NewManager(capacity int, baseDir string) DropManager {
	msgRing := ring.New(capacity)
	return DropManager{
		capacity:   capacity,
		baseDir:    baseDir,
		msgRing:    msgRing,
		submitChan: launchRingManager(msgRing),
	}
}

func launchRingManager(msgRing *ring.Ring) chan message {
	submitChan := make(chan message)
	go func() {
		for {
			msg := <-submitChan
			oldMsg, oldExisted := msgRing.Value.(message)
			msgRing.Value = msg
			if oldExisted {
				err := oldMsg.discard()
				if err != nil {
					log.Panic("Failed to discard message")
				}
			}
			msgRing = msgRing.Next()
		}
	}()
	return submitChan
}

func (server *DropManager) Submit(drop Drop, data io.Reader) error {
	msg := message{
		drop:      drop,
		timestamp: time.Now(),
	}
	file, err := ioutil.TempFile(server.baseDir, "")
	if err != nil {
		return err
	}
	defer file.Close()
	msg.path = file.Name()
	_, err = io.CopyN(file, data, MaximumMessageSize+1)
	if err != io.EOF {
		msg.discard()
		if err == nil {
			return MessageSizeExceededError{}
		} else {
			// some other unforeseen error occurred
			return err<commit_msg>Prefixed message file paths with drop id<commit_after>package drop

import (
	"container/ring"
	"encoding/base64"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"os"
	"time"
)

const (
	MaximumMessageSize = 4096
	dropIdLength       = 32
)

type MessageSizeExceededError struct{}

func (m MessageSizeExceededError) Error() string {
	return fmt.Sprintf("drop: maximum message size of %d bytes exceeded", MaximumMessageSize)
}

type Drop struct {
	Id string
}

func (d Drop) Verify() bool {
	// must be 32 byte encoded to 43 base64url characters
	bytes, err := base64.URLEncoding.DecodeString(d.Id + "=") // add padding to make parser happy
	return err == nil && len(bytes) == dropIdLength
}

type message struct {
	drop      Drop
	timestamp time.Time
	path      string
}

func (msg *message) prefix() string {
	return fmt.Sprintf("%v_", msg.drop.Id)
}

func (msg *message) discard() error {
	return os.Remove(msg.path)
}

type DropManager struct {
	capacity   int
	baseDir    string
	msgRing    *ring.Ring
	submitChan chan message
}

func NewManager(capacity int, baseDir string) DropManager {
	msgRing := ring.New(capacity)
	return DropManager{
		capacity:   capacity,
		baseDir:    baseDir,
		msgRing:    msgRing,
		submitChan: launchRingManager(msgRing),
	}
}

func launchRingManager(msgRing *ring.Ring) chan message {
	submitChan := make(chan message)
	go func() {
		for {
			msg := <-submitChan
			oldMsg, oldExisted := msgRing.Value.(message)
			msgRing.Value = msg
			if oldExisted {
				err := oldMsg.discard()
				if err != nil {
					log.Panic("Failed to discard message")
				}
			}
			msgRing = msgRing.Next()
		}
	}()
	return submitChan
}

func (server *DropManager) Submit(drop Drop, data io.Reader) error {
	msg := message{
		drop:      drop,
		timestamp: time.Now(),
	}
	file, err := ioutil.TempFile(server.baseDir, msg.prefix())
	if err != nil {
		return err
	}
	defer file.Close()
	msg.path = file.Name()
	_, err = io.CopyN(file, data, MaximumMessageSize+1)
	if err != io.EOF {
		msg.discard()
		if err == nil {
			return MessageSizeExceededError{}
		} else {
			// some other unforeseen error occurred
			return err
{'fraction_non_alphanumeric': 0.11674757281553398, 'fraction_numerical': 0.00825242718446602, 'mean_word_length': 3.474484256243214, 'pattern_counts': {'":': 0, '<': 5, '<?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': '7503764', 'n_tokens_mistral': 1575, 'n_tokens_neox': 1404, 'n_words': 400}