<commit_before>//
// Copyright © 2011-2016 <NAME>
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//

package stompngo

import (
	"strconv"
	"strings"
	"time"
)

/*
	Logical network reader.

	Read STOMP frames from the connection, create MessageData
	structures from the received data, and push the MessageData to the client.
*/
func (c *Connection) reader() {
readLoop:
	for {
		f, e := c.readFrame()
		c.log("RDR_RECEIVE_FRAME", f.Command, f.Headers, hexData(f.Body),
			"RDR_RECEIVE_ERR", e)
		if e != nil {
			f.Headers = append(f.Headers, "connection_read_error", e.Error())
			md := MessageData{Message(f), e}
			c.handleReadError(md)
			c.log("RDR_CONN_ERR", e)
			break readLoop
		}

		// if f.Command == "" {
		//	break
		//}

		m := Message(f)
		c.mets.tfr += 1 // Total frames read
		// Headers already decoded
		c.mets.tbr += m.Size(false) // Total bytes read

		// TODO START - can this be simplified ?  Look cleaner ?
		md := MessageData{m, e}
		if sid, ok := f.Headers.Contains(HK_SUBSCRIPTION); ok {
			// This is a read lock
			c.subsLock.RLock()
			// This sub can be already gone under some timing circumstances
			if _, sok := c.subs[sid]; sok {
				// And it can also be closed under some timing circumstances
				if c.subs[sid].cs {
					c.log("RDR_CLSUB", sid, m.Command, m.Headers)
				} else {
					if c.subs[sid].drav {
						c.subs[sid].drmc++
						if c.subs[sid].drmc > c.subs[sid].dra {
							c.log("RDR_DROPM", c.subs[sid].drmc, sid, m.Command,
								m.Headers, hexData(m.Body))
						} else {
							c.subs[sid].md <- md
						}
					} else {
						c.subs[sid].md <- md
					}
				}
			} else {
				c.log("RDR_NOSUB", sid, m.Command, m.Headers)
			}
			c.subsLock.RUnlock()
		} else {
			// RECEIPTs and ERRORs are never drained.  They actually cannot
			// be drained in any logical manner because they do not have a
			// 'subscription' header.
			c.input <- md
		}

		// TODO END

		select {
		case _ = <-c.ssdc:
			c.log("RDR_SHUTDOWN detected")
			break readLoop
		default:
		}
		c.log("RDR_RELOOP")
	}
	close(c.input)
	c.log("RDR_SHUTDOWN", time.Now())
}

/*
	Physical frame reader.

	This parses a single STOMP frame from data off of the wire, and
	returns a Frame, with a possible error.

	Note: this functionality could hang or exhibit other erroneous behavior
	if running against a non-compliant STOMP server.
*/
func (c *Connection) readFrame() (f Frame, e error) {
	f = Frame{"", Headers{}, NULLBUFF}
	// Read f.Command or line ends (maybe heartbeats)
	for {
		s, e := c.rdr.ReadString('\n')
		if s == "" {
			return f, e<commit_msg>Restructure reader logic in main loop:<commit_after>//
// Copyright © 2011-2016 <NAME>
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//

package stompngo

import (
	"fmt"
	"strconv"
	"strings"
	"time"
)

/*
	Logical network reader.

	Read STOMP frames from the connection, create MessageData
	structures from the received data, and push the MessageData to the client.
*/
func (c *Connection) reader() {
readLoop:
	for {
		f, e := c.readFrame()
		c.log("RDR_RECEIVE_FRAME", f.Command, f.Headers, hexData(f.Body),
			"RDR_RECEIVE_ERR", e)
		if e != nil {
			f.Headers = append(f.Headers, "connection_read_error", e.Error())
			md := MessageData{Message(f), e}
			c.handleReadError(md)
			c.log("RDR_CONN_ERR", e)
			break readLoop
		}

		// if f.Command == "" {
		//	break
		//}

		m := Message(f)
		c.mets.tfr += 1 // Total frames read
		// Headers already decoded
		c.mets.tbr += m.Size(false) // Total bytes read

		//*************************************************************************
		// Replacement START
		md := MessageData{m, e}
		switch f.Command {
		//
		case MESSAGE:
			sid, ok := f.Headers.Contains(HK_SUBSCRIPTION)
			if !ok { // This should *NEVER* happen
				panic(fmt.Sprintf("stompngo INTERNAL ERROR: command:<%s> headers:<%v>",
					f.Command, f.Headers))
			}
			c.subsLock.RLock()
			ps, sok := c.subs[sid] // This is a map of pointers .....
			//
			if !sok {
				// The sub can be gone under some timing conditions.  In that case
				// we log it of possible, and continue (hope for the best).
				c.log("RDR_NOSUB", sid, m.Command, m.Headers)
				goto csRUnlock
			}
			if ps.cs {
				// The sub can also already be closed under some conditions.
				// Again, we log that if possible, and continue
				c.log("RDR_CLSUB", sid, m.Command, m.Headers)
				goto csRUnlock
			}
			// Handle subscription draining
			switch ps.drav {
			case false:
				ps.md <- md
			default:
				ps.drmc++
				if ps.drmc > ps.dra {
					c.log("RDR_DROPM", ps.drmc, sid, m.Command,
						m.Headers, hexData(m.Body))
				} else {
					ps.md <- md
				}
			}
		csRUnlock:
			c.subsLock.RUnlock()
		//
		case ERROR:
			fallthrough
		//
		case RECEIPT:
			c.input <- md
		//
		default:
			panic(fmt.Sprintf("Broker SEVERE ERROR, not STOMP? command:<%s> headers:<%v>",
				f.Command, f.Headers))
		}
		// Replacement END
		//*************************************************************************

		select {
		case _ = <-c.ssdc:
			c.log("RDR_SHUTDOWN detected")
			break readLoop
		default:
		}
		c.log("RDR_RELOOP")
	}
	close(c.input)
	c.log("RDR_SHUTDOWN", time.Now())
}

/*
	Physical frame reader.

	This parses a single STOMP frame from data off of the wire, and
	returns a Frame, with a possible error.

	Note: this functionality could hang or exhibit other erroneous behavior
	if running against a non-compliant STOMP server.
*/
func (c *Connection) readFrame() (f Frame, e error) {
	f = Frame{"", Headers{}, NULLBUFF}
	// Read f.Command or line ends (maybe heartbeats)
	for {
		s, e := c.rdr.ReadString('\n')
		if s == "" {
			return f, e
{'fraction_non_alphanumeric': 0.15063138905592302, 'fraction_numerical': 0.0039085989176187615, 'mean_word_length': 2.8345821325648415, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '3155503', 'n_tokens_mistral': 2625, 'n_tokens_neox': 2244, 'n_words': 727}