<commit_before>package apns

import (
	"container/list"
	"crypto/tls"
	"encoding/binary"
	"encoding/hex"
	"fmt"
	"strings"
	"sync"
	"time"
)

func newPushService() (*pushService, error) {
	ps := &pushService{}
	if err := ps.createConnection(); err != nil {
		return nil, err
	}

	ps.waitClose = &sync.WaitGroup{}
	ps.waitClose.Add(3)

	ps.sentQueue = make(chan *ApnsMessage, 1000)
	ps.confirmedQueue = make(chan *apnsReceiveMessage, 1000)
	ps.doSenderClose = make(chan bool, 2)

	go ps.monitorReply()
	go ps.monitorMessages()
	go ps.messageSender()

	return ps, nil
}

type pushService struct {
	conn      *tls.Conn
	lastError error

	sentQueue      chan *ApnsMessage
	confirmedQueue chan *apnsReceiveMessage

	doSenderClose chan bool
	waitClose     *sync.WaitGroup
}

func (ps *pushService) createConnection() error {
	conf, err := getTlsConfig()
	if err != nil {
		return err
	}

	conn, err := tls.Dial("tcp", defaultConfig.gateway, conf)
	if err != nil {
		return err
	}

	ps.conn = conn
	return nil
}

func (ps *pushService) destroy() {
	logdebug("Destroy push service %p, waiting to finish", ps)

	ps.doSenderClose <- true //this will close the sender
	ps.conn.Close()          //this will close the socket and force the read to exit
	//(will remove the last message if not confirmed ??? it's ok ???)
	//when sender/reader are closed the sending monitor will close

	ps.waitClose.Wait()
	logdebug("Destroy push service %p finish", ps)
}

func (ps *pushService) messageSender() {
	defer func() {
		ps.waitClose.Done()
		logdebug("Message sender stopped %p", ps)
	}()

	logdebug("Message sender started %p", ps)

	for {
		select {
		case <-ps.doSenderClose:
			close(ps.sentQueue)
			return
		case messages := <-mainSendingMessage:
			logdebug("Sending new message: %d to %s", messages.id, messages.DeviceToken)
			data, err := createBinaryNotification(messages, messages.id, messages.DeviceToken)
			if err != nil {
				messages.SetStatus(err, true)
				break
			}

			_, err = ps.conn.Write(data)
			if err != nil {
				ps.lastError = nil
				pushPool.releasedServices <- ps
				close(ps.sentQueue)
				return
			}
			logdebug("Message id: %d sent", messages.id)
			ps.sentQueue <- messages
			pushPool.reportQueue <- &notifReport{sent: 1}
		}
	}
}

func (ps *pushService) monitorReply() {
	defer func() {
		ps.waitClose.Done()
		logdebug("Message reply stopped %p", ps)
	}()

	logdebug("Message reply started %p", ps)

	readb := [6]byte{}

	for {
		n, err := ps.conn.Read(readb[:])

		if err != nil {
			logerr("APNS read channel is closed %s", err)
			//when the read socket channel is closed the last payload i think isn't sent ok
			//i don't know the last message id so we consider id 0 as last message and don't use this socket
			ps.lastError = err
			ps.confirmedQueue <- &apnsReceiveMessage{lastMessage: true}
			ps.doSenderClose <- true
			close(ps.confirmedQueue)
			break
		}

		if n == 6 {
			// if status 0 (this i think isn't going to happen as apple says)
			// if status 2 to 8 it's device token or payload error, don't try to resend
			// if status 1 it's a proccessing error, we should retry sending the payload
			// if status 10 the payload was sent but we must not send anything else on this socket (is shutdown)
			// if status 255 it's unknown error (let's retry to send the payload)  - close socket
			// if status number unknown ???? (ok, let's retry to send the payload) - close socket

			status := readb[1]
			id := binary.BigEndian.Uint32(readb[2:])

			ps.confirmedQueue <- &apnsReceiveMessage{status: status, id: id}
			logdebug("Received confirmation for id: %d with status: %d", id, status)
			if status >= 9 {
				ps.lastError = fmt.Errorf("APNs server reply with status %d (%s), closing", status, statusToError(status))
				ps.doSenderClose <- true
				close(ps.confirmedQueue)
				break
			}
		} else {
			//unknow data sent from apple
			// let's close the socket and mark all the messagess with error
			logerr("Unknow apple message (%s) socket will be closed", hex.EncodeToString(readb[:n]))
			//we consinder that max uint32 means to delete all the messages from the queue
			ps.lastError = fmt.Errorf("Unknow apple message (%s) socket will be closed", hex.EncodeToString(readb[:n]))
			ps.confirmedQueue <- &apnsReceiveMessage{unknownMessage: true}
			ps.doSenderClose <- true
			close(ps.confirmedQueue)
			break<commit_msg>Don't send messages on a connection that had an error<commit_after>package apns

import (
	"container/list"
	"crypto/tls"
	"encoding/binary"
	"encoding/hex"
	"fmt"
	"strings"
	"sync"
	"time"
)

func newPushService() (*pushService, error) {
	ps := &pushService{}
	if err := ps.createConnection(); err != nil {
		return nil, err
	}

	ps.waitClose = &sync.WaitGroup{}
	ps.waitClose.Add(3)

	ps.sentQueue = make(chan *ApnsMessage, 1000)
	ps.confirmedQueue = make(chan *apnsReceiveMessage, 1000)
	ps.doSenderClose = make(chan bool, 2)

	go ps.monitorReply()
	go ps.monitorMessages()
	go ps.messageSender()

	return ps, nil
}

type pushService struct {
	conn      *tls.Conn
	lastError error

	sentQueue      chan *ApnsMessage
	confirmedQueue chan *apnsReceiveMessage

	doSenderClose chan bool
	waitClose     *sync.WaitGroup
}

func (ps *pushService) createConnection() error {
	conf, err := getTlsConfig()
	if err != nil {
		return err
	}

	conn, err := tls.Dial("tcp", defaultConfig.gateway, conf)
	if err != nil {
		return err
	}

	ps.conn = conn
	return nil
}

func (ps *pushService) destroy() {
	logdebug("Destroy push service %p, waiting to finish", ps)

	ps.doSenderClose <- true //this will close the sender
	ps.conn.Close()          //this will close the socket and force the read to exit
	//(will remove the last message if not confirmed ??? it's ok ???)
	//when sender/reader are closed the sending monitor will close

	ps.waitClose.Wait()
	logdebug("Destroy push service %p finish", ps)
}

func (ps *pushService) messageSender() {
	defer func() {
		ps.waitClose.Done()
		logdebug("Message sender stopped %p", ps)
	}()

	logdebug("Message sender started %p", ps)

	for {
		select {
		case <-ps.doSenderClose:
			close(ps.sentQueue)
			return
		case messages := <-mainSendingMessage:
			logdebug("Sending new message: %d to %s", messages.id, messages.DeviceToken)
			data, err := createBinaryNotification(messages, messages.id, messages.DeviceToken)
			if err != nil {
				messages.SetStatus(err, true)
				break
			}

			_, err = ps.conn.Write(data)
			if err != nil {
				ps.lastError = nil
				pushPool.releasedServices <- ps
				close(ps.sentQueue)
				return
			}
			logdebug("Message id: %d sent", messages.id)
			ps.sentQueue <- messages
			pushPool.reportQueue <- &notifReport{sent: 1}
		}
	}
}

func (ps *pushService) monitorReply() {
	defer func() {
		ps.waitClose.Done()
		logdebug("Message reply stopped %p", ps)
	}()

	logdebug("Message reply started %p", ps)

	readb := [6]byte{}

	for {
		n, err := ps.conn.Read(readb[:])

		if err != nil {
			logerr("APNS read channel is closed %s", err)
			//when the read socket channel is closed the last payload i think isn't sent ok
			//i don't know the last message id so we consider id 0 as last message and don't use this socket
			ps.lastError = err
			ps.confirmedQueue <- &apnsReceiveMessage{lastMessage: true}
			ps.doSenderClose <- true
			close(ps.confirmedQueue)
			break
		}

		if n == 6 {
			// if status 0 (this i think isn't going to happen as apple says)
			// if status 2 to 8 it's device token or payload error, don't try to resend
			// if status 1 it's a proccessing error, we should retry sending the payload
			// if status 10 the payload was sent but we must not send anything else on this socket (is shutdown)
			// if status 255 it's unknown error (let's retry to send the payload)  - close socket
			// if status number unknown ???? (ok, let's retry to send the payload) - close socket

			status := readb[1]
			id := binary.BigEndian.Uint32(readb[2:])

			ps.confirmedQueue <- &apnsReceiveMessage{status: status, id: id}
			logdebug("Received confirmation for id: %d with status: %d", id, status)
			if status != 0 {
				ps.lastError = fmt.Errorf("APNs server reply with status %d (%s), closing", status, statusToError(status))
				ps.doSenderClose <- true
				close(ps.confirmedQueue)
				break
			}
		} else {
			//unknow data sent from apple
			// let's close the socket and mark all the messagess with error
			logerr("Unknow apple message (%s) socket will be closed", hex.EncodeToString(readb[:n]))
			//we consinder that max uint32 means to delete all the messages from the queue
			ps.lastError = fmt.Errorf("Unknow apple message (%s) socket will be closed", hex.EncodeToString(readb[:n]))
			ps.confirmedQueue <- &apnsReceiveMessage{unknownMessage: true}
			ps.doSenderClose <- true
			close(ps.confirmedQueue)
			break
{'fraction_non_alphanumeric': 0.11668943533697632, 'fraction_numerical': 0.006830601092896175, 'mean_word_length': 3.4913087934560325, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 4, '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': '5658197', 'n_tokens_mistral': 3208, 'n_tokens_neox': 2776, 'n_words': 964}