<commit_before>package nats

import (
	"testing"
	"fmt"
//	"os"
	"os/exec"
	"net"
	"time"
	"math"
	"runtime"
	"regexp"
	"bytes"
	"strings"
)

const natsServer = "nats-server"

type server struct {
	args []string
	cmd  *exec.Cmd
}

var s *server

func startNatsServer(t *testing.T, port uint, other string) *server {
	var s server
	args := fmt.Sprintf("-p %d %s", port, other)
	s.args = strings.Split(args, " ")
	s.cmd = exec.Command(natsServer, s.args...)
	err := s.cmd.Start()
	if err != nil {
		s.cmd = nil
		t.Errorf("Could not start %s, is NATS installed and in path?", natsServer)
		return &s
	}
	// Give it time to start up
	start := time.Now()
	for {
		addr := fmt.Sprintf("localhost:%d", port)
		c, err := net.Dial("tcp", addr)
		if err != nil {
			time.Sleep(50 * time.Millisecond)
			if time.Since(start) > (5 * time.Second) {
				t.Fatalf("Timed out trying to connect to %s", natsServer)
				return nil
			}
		} else {
			c.Close()
			break
		}
	}
	return &s
}

func (s *server) stopServer() {
	if s.cmd != nil && s.cmd.Process != nil {
		s.cmd.Process.Kill()
	}
}

func newConnection(t *testing.T) Connection {
	nc, err := DefaultConnection()
	if err != nil {
		t.Fatal("Failed to create default connection", err)
		return nil
	}
	return nc
}

func TestDefaultConnection(t *testing.T) {
	s = startNatsServer(t, DefaultPort, "")
	nc := newConnection(t)
	nc.Close()
}

func TestClose(t *testing.T) {
	base := runtime.NumGoroutine()
	nc := newConnection(t)
	time.Sleep(10 * time.Millisecond)
	nc.Close()
	time.Sleep(10 * time.Millisecond)
	delta := (runtime.NumGoroutine() - base)
	if delta > 0 {
		t.Fatalf("%d Go routines still exists post Close()", delta)
	}
	// Make sure we can call Close() multiple times
	nc.Close()
}

func TestSimplePublish(t *testing.T) {
	nc := newConnection(t)
	defer nc.Close()
	err := nc.Publish("foo", []byte("Hello World"))
	if err != nil {
		t.Fatal("Failed to publish string message: ", err)
	}
}

func TestAsyncSubscribe(t *testing.T) {
	nc := newConnection(t)
	defer nc.Close()
	omsg := []byte("Hello World")
	received := 0
	_, err := nc.Subscribe("foo", func(subj, _ string, data []byte, sub *Subscription) {
		received += 1
		if !bytes.Equal(data, omsg) {
			t.Fatal("Message received does not match")
		}
		if (sub == nil) {
			t.Fatal("Callback does not have a valid Subsription")
		}
	})
	if err != nil {
		t.Fatal("Failed to subscribe: ", err)
	}
	nc.Publish("foo", omsg)
	nc.Flush()
	if (received != 1) {
		t.Fatal("Message not received for subscription")
	}
}

func TestSyncSubscribe(t *testing.T) {
	nc := newConnection(t)
	defer nc.Close()<commit_msg>Added empty msg publish test<commit_after>package nats

import (
	"testing"
	"fmt"
//	"os"
	"os/exec"
	"net"
	"time"
	"math"
	"runtime"
	"regexp"
	"bytes"
	"strings"
)

const natsServer = "nats-server"

type server struct {
	args []string
	cmd  *exec.Cmd
}

var s *server

func startNatsServer(t *testing.T, port uint, other string) *server {
	var s server
	args := fmt.Sprintf("-p %d %s", port, other)
	s.args = strings.Split(args, " ")
	s.cmd = exec.Command(natsServer, s.args...)
	err := s.cmd.Start()
	if err != nil {
		s.cmd = nil
		t.Errorf("Could not start %s, is NATS installed and in path?", natsServer)
		return &s
	}
	// Give it time to start up
	start := time.Now()
	for {
		addr := fmt.Sprintf("localhost:%d", port)
		c, err := net.Dial("tcp", addr)
		if err != nil {
			time.Sleep(50 * time.Millisecond)
			if time.Since(start) > (5 * time.Second) {
				t.Fatalf("Timed out trying to connect to %s", natsServer)
				return nil
			}
		} else {
			c.Close()
			break
		}
	}
	return &s
}

func (s *server) stopServer() {
	if s.cmd != nil && s.cmd.Process != nil {
		s.cmd.Process.Kill()
	}
}

func newConnection(t *testing.T) Connection {
	nc, err := DefaultConnection()
	if err != nil {
		t.Fatal("Failed to create default connection", err)
		return nil
	}
	return nc
}

func TestDefaultConnection(t *testing.T) {
	s = startNatsServer(t, DefaultPort, "")
	nc := newConnection(t)
	nc.Close()
}

func TestClose(t *testing.T) {
	base := runtime.NumGoroutine()
	nc := newConnection(t)
	time.Sleep(10 * time.Millisecond)
	nc.Close()
	time.Sleep(10 * time.Millisecond)
	delta := (runtime.NumGoroutine() - base)
	if delta > 0 {
		t.Fatalf("%d Go routines still exists post Close()", delta)
	}
	// Make sure we can call Close() multiple times
	nc.Close()
}

func TestSimplePublish(t *testing.T) {
	nc := newConnection(t)
	defer nc.Close()
	err := nc.Publish("foo", []byte("Hello World"))
	if err != nil {
		t.Fatal("Failed to publish string message: ", err)
	}
}

func TestSimplePublishNoData(t *testing.T) {
	nc := newConnection(t)
	defer nc.Close()
	err := nc.Publish("foo", nil)
	if err != nil {
		t.Fatal("Failed to publish empty message: ", err)
	}
}

func TestAsyncSubscribe(t *testing.T) {
	nc := newConnection(t)
	defer nc.Close()
	omsg := []byte("Hello World")
	received := 0
	_, err := nc.Subscribe("foo", func(subj, _ string, data []byte, sub *Subscription) {
		received += 1
		if !bytes.Equal(data, omsg) {
			t.Fatal("Message received does not match")
		}
		if (sub == nil) {
			t.Fatal("Callback does not have a valid Subsription")
		}
	})
	if err != nil {
		t.Fatal("Failed to subscribe: ", err)
	}
	nc.Publish("foo", omsg)
	nc.Flush()
	if (received != 1) {
		t.Fatal("Message not received for subscription")
	}
}

func TestSyncSubscribe(t *testing.T) {
	nc := newConnection(t)
	defer nc.Close()
{'fraction_non_alphanumeric': 0.16420664206642066, 'fraction_numerical': 0.004059040590405904, 'mean_word_length': 3.5175, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '1296167', 'n_tokens_mistral': 2184, 'n_tokens_neox': 2061, 'n_words': 558}