func (s *TxStoreSuite) TestSuccessfulTx(c *C) {
	txs := txStore{}

	err := txs.Begin("tx1")
	c.Check(err, IsNil)

	err = txs.Begin("tx2")
	c.Assert(err, IsNil)

	f1 := frame.New(frame.MESSAGE,
		frame.Destination, "/queue/1")

	f2 := frame.New(frame.MESSAGE,
		frame.Destination, "/queue/2")

	f3 := frame.New(frame.MESSAGE,
		frame.Destination, "/queue/3")

	f4 := frame.New(frame.MESSAGE,
		frame.Destination, "/queue/4")

	err = txs.Add("tx1", f1)
	c.Assert(err, IsNil)
	err = txs.Add("tx1", f2)
	c.Assert(err, IsNil)
	err = txs.Add("tx1", f3)
	c.Assert(err, IsNil)
	err = txs.Add("tx2", f4)

	var tx1 []*frame.Frame

	txs.Commit("tx1", func(f *frame.Frame) error {
		tx1 = append(tx1, f)
		return nil
	})
	c.Check(err, IsNil)

	var tx2 []*frame.Frame

	err = txs.Commit("tx2", func(f *frame.Frame) error {
		tx2 = append(tx2, f)
		return nil
	})
	c.Check(err, IsNil)

	c.Check(len(tx1), Equals, 3)
	c.Check(tx1[0], Equals, f1)
	c.Check(tx1[1], Equals, f2)
	c.Check(tx1[2], Equals, f3)

	c.Check(len(tx2), Equals, 1)
	c.Check(tx2[0], Equals, f4)

	// already committed, so should cause an error
	err = txs.Commit("tx1", func(f *frame.Frame) error {
		c.Fatal("should not be called")
		return nil
	})
	c.Check(err, Equals, txUnknown)
}

func (s *MemoryQueueSuite) Test1(c *C) {
	mq := NewMemoryQueueStorage()
	mq.Start()

	f1 := frame.New(frame.MESSAGE,
		frame.Destination, "/queue/test",
		frame.MessageId, "msg-001",
		frame.Subscription, "1")

	err := mq.Enqueue("/queue/test", f1)
	c.Assert(err, IsNil)

	f2 := frame.New(frame.MESSAGE,
		frame.Destination, "/queue/test",
		frame.MessageId, "msg-002",
		frame.Subscription, "1")

	err = mq.Enqueue("/queue/test", f2)
	c.Assert(err, IsNil)

	f3 := frame.New(frame.MESSAGE,
		frame.Destination, "/queue/test2",
		frame.MessageId, "msg-003",
		frame.Subscription, "2")

	err = mq.Enqueue("/queue/test2", f3)
	c.Assert(err, IsNil)

	// attempt to dequeue from a different queue
	f, err := mq.Dequeue("/queue/other-queue")
	c.Check(err, IsNil)
	c.Assert(f, IsNil)

	f, err = mq.Dequeue("/queue/test2")
	c.Check(err, IsNil)
	c.Assert(f, Equals, f3)

	f, err = mq.Dequeue("/queue/test")
	c.Check(err, IsNil)
	c.Assert(f, Equals, f1)

	f, err = mq.Dequeue("/queue/test")
	c.Check(err, IsNil)
	c.Assert(f, Equals, f2)

	f, err = mq.Dequeue("/queue/test")
	c.Check(err, IsNil)
	c.Assert(f, IsNil)

	f, err = mq.Dequeue("/queue/test2")
	c.Check(err, IsNil)
	c.Assert(f, IsNil)
}

func (s *StompSuite) Test_successful_connect_with_nonstandard_header(c *C) {
	resetId()
	fc1, fc2 := testutil.NewFakeConn(c)
	stop := make(chan struct{})

	go func() {
		defer func() {
			fc2.Close()
			close(stop)
		}()
		reader := frame.NewReader(fc2)
		writer := frame.NewWriter(fc2)

		f1, err := reader.Read()
		c.Assert(err, IsNil)
		c.Assert(f1.Command, Equals, "CONNECT")
		c.Assert(f1.Header.Get("login"), Equals, "guest")
		c.Assert(f1.Header.Get("passcode"), Equals, "guest")
		c.Assert(f1.Header.Get("host"), Equals, "/")
		c.Assert(f1.Header.Get("x-max-length"), Equals, "50")
		connectedFrame := frame.New("CONNECTED")
		connectedFrame.Header.Add("session", "session-0voRHrG-VbBedx1Gwwb62Q")
		connectedFrame.Header.Add("heart-beat", "0,0")
		connectedFrame.Header.Add("server", "RabbitMQ/3.2.1")
		connectedFrame.Header.Add("version", "1.0")
		writer.Write(connectedFrame)

		f2, err := reader.Read()
		c.Assert(err, IsNil)
		c.Assert(f2.Command, Equals, "DISCONNECT")
		receipt, _ := f2.Header.Contains("receipt")
		c.Check(receipt, Equals, "1")

		writer.Write(frame.New("RECEIPT", frame.ReceiptId, "1"))
	}()

	client, err := Connect(fc1,
		ConnOpt.Login("guest", "guest"),
		ConnOpt.Host("/"),
		ConnOpt.Header("x-max-length", "50"))
	c.Assert(err, IsNil)
	c.Assert(client, NotNil)
	c.Assert(client.Version(), Equals, V10)
	c.Assert(client.Session(), Equals, "session-0voRHrG-VbBedx1Gwwb62Q")
	c.Assert(client.Server(), Equals, "RabbitMQ/3.2.1")

	err = client.Disconnect()
	c.Assert(err, IsNil)

	<-stop
}

func (s *FrameSuite) TestDetermineVersion_MultipleVersions(c *C) {
	f := frame.New(frame.CONNECT)
	f.Header.Add(frame.AcceptVersion, "1.2,1.1,1.0,2.0")
	version, err := determineVersion(f)
	c.Check(version, Equals, stomp.V12)
	c.Check(err, IsNil)
}

func (s *FrameSuite) TestDetermineVersion_V11_Connect(c *C) {
	f := frame.New(frame.CONNECT)
	f.Header.Add(frame.AcceptVersion, "1.1")
	version, err := determineVersion(f)
	c.Check(version, Equals, stomp.V11)
	c.Check(err, IsNil)
}

func (s *FrameSuite) TestDetermineVersion_V10_Stomp(c *C) {
	// the "STOMP" command was introduced in V1.1, so it must
	// have an accept-version header
	f := frame.New(frame.STOMP)
	_, err := determineVersion(f)
	c.Check(err, Equals, missingHeader(frame.AcceptVersion))
}

func (co *connOptions) NewFrame() (*frame.Frame, error) {
	f := frame.New(co.FrameCommand)
	if co.Host != "" {
		f.Header.Set(frame.Host, co.Host)
	}

	// heart-beat
	{
		send := co.WriteTimeout / time.Millisecond
		recv := co.ReadTimeout / time.Millisecond
		f.Header.Set(frame.HeartBeat, fmt.Sprintf("%d,%d", send, recv))
	}

	// login, passcode
	if co.Login != "" || co.Passcode != "" {
		f.Header.Set(frame.Login, co.Login)
		f.Header.Set(frame.Passcode, co.Passcode)
	}

	// accept-version
	f.Header.Set(frame.AcceptVersion, strings.Join(co.AcceptVersions, ","))

	// custom header entries -- note that these do not override
	// header values already set as they are added to the end of
	// the header array
	f.Header.AddHeader(co.Header)

	return f, nil
}

func (s *FrameSuite) TestHeartBeat(c *C) {
	f := frame.New(frame.CONNECT,
		frame.AcceptVersion, "1.2",
		frame.Host, "XX")

	// no heart-beat header means zero values
	x, y, err := getHeartBeat(f)
	c.Check(x, Equals, 0)
	c.Check(y, Equals, 0)
	c.Check(err, IsNil)

	f.Header.Add("heart-beat", "123,456")
	x, y, err = getHeartBeat(f)
	c.Check(x, Equals, 123)
	c.Check(y, Equals, 456)
	c.Check(err, IsNil)

	f.Header.Set(frame.HeartBeat, "invalid")
	x, y, err = getHeartBeat(f)
	c.Check(x, Equals, 0)
	c.Check(y, Equals, 0)
	c.Check(err, Equals, invalidHeartBeat)

	f.Header.Del(frame.HeartBeat)
	_, _, err = getHeartBeat(f)
	c.Check(err, IsNil)

	f.Command = frame.SEND
	_, _, err = getHeartBeat(f)
	c.Check(err, Equals, invalidOperationForFrame)
}

// Unsubscribes and closes the channel C.
func (s *Subscription) Unsubscribe(opts ...func(*frame.Frame) error) error {
	if s.completed {
		return ErrCompletedSubscription
	}
	f := frame.New(frame.UNSUBSCRIBE, frame.Id, s.id)

	for _, opt := range opts {
		if opt == nil {
			return ErrNilOption
		}
		err := opt(f)
		if err != nil {
			return err
		}
	}

	s.conn.sendFrame(f)
	s.completedMutex.Lock()
	if !s.completed {
		s.completed = true
		close(s.C)
	}
	s.completedMutex.Unlock()
	return nil
}

// Sets up a connection for testing
func connectHelper(c *C, version Version) (*Conn, *fakeReaderWriter) {
	fc1, fc2 := testutil.NewFakeConn(c)
	stop := make(chan struct{})

	reader := frame.NewReader(fc2)
	writer := frame.NewWriter(fc2)

	go func() {
		f1, err := reader.Read()
		c.Assert(err, IsNil)
		c.Assert(f1.Command, Equals, "CONNECT")
		f2 := frame.New("CONNECTED", "version", version.String())
		writer.Write(f2)
		close(stop)
	}()

	conn, err := Connect(fc1)
	c.Assert(err, IsNil)
	c.Assert(conn, NotNil)
	<-stop
	return conn, &fakeReaderWriter{
		reader: reader,
		writer: writer,
		conn:   fc2,
	}
}

func (s *StompSuite) TestHeartBeatReadTimeout(c *C) {
	conn, rw := createHeartBeatConnection(c, 100, 10000, time.Millisecond)

	go func() {
		f1, err := rw.Read()
		c.Assert(err, IsNil)
		c.Assert(f1.Command, Equals, "SUBSCRIBE")
		messageFrame := frame.New("MESSAGE",
			"destination", f1.Header.Get("destination"),
			"message-id", "1",
			"subscription", f1.Header.Get("id"))
		messageFrame.Body = []byte("Message body")
		rw.Write(messageFrame)
	}()

	sub, err := conn.Subscribe("/queue/test1", AckAuto)
	c.Assert(err, IsNil)
	c.Check(conn.readTimeout, Equals, 101*time.Millisecond)
	//println("read timeout", conn.readTimeout.String())

	msg, ok := <-sub.C
	c.Assert(msg, NotNil)
	c.Assert(ok, Equals, true)

	msg, ok = <-sub.C
	c.Assert(msg, NotNil)
	c.Assert(msg.Err, NotNil)
	c.Assert(msg.Err.Error(), Equals, "read timeout")

	msg, ok = <-sub.C
	c.Assert(msg, IsNil)
	c.Assert(ok, Equals, false)
}

func createHeartBeatConnection(
	c *C,
	readTimeout, writeTimeout int,
	readTimeoutError time.Duration) (*Conn, *fakeReaderWriter) {
	fc1, fc2 := testutil.NewFakeConn(c)
	stop := make(chan struct{})

	reader := frame.NewReader(fc2)
	writer := frame.NewWriter(fc2)

	go func() {
		f1, err := reader.Read()
		c.Assert(err, IsNil)
		c.Assert(f1.Command, Equals, "CONNECT")
		c.Assert(f1.Header.Get("heart-beat"), Equals, "1,1")
		f2 := frame.New("CONNECTED", "version", "1.2")
		f2.Header.Add("heart-beat", fmt.Sprintf("%d,%d", readTimeout, writeTimeout))
		writer.Write(f2)
		close(stop)
	}()

	conn, err := Connect(fc1,
		ConnOpt.HeartBeat(time.Millisecond, time.Millisecond),
		ConnOpt.HeartBeatError(readTimeoutError))
	c.Assert(conn, NotNil)
	c.Assert(err, IsNil)
	<-stop
	return conn, &fakeReaderWriter{
		reader: reader,
		writer: writer,
		conn:   fc2,
	}
}

func (s *FrameSuite) TestDetermineVersion_IncompatibleVersions(c *C) {
	f := frame.New(frame.CONNECT)
	f.Header.Add(frame.AcceptVersion, "0.2,0.1,1.3,2.0")
	version, err := determineVersion(f)
	c.Check(version, Equals, stomp.Version(""))
	c.Check(err, Equals, unknownVersion)
}

func (s *TopicSuite) TestTopicWithoutSubscription(c *C) {
	topic := newTopic("destination")

	f := frame.New(frame.MESSAGE,
		frame.Destination, "destination")

	topic.Enqueue(f)
}

// Create an ACK or NACK frame. Complicated by version incompatibilities.
func (c *Conn) createAckNackFrame(msg *Message, ack bool) (*frame.Frame, error) {
	if !ack && !c.version.SupportsNack() {
		return nil, ErrNackNotSupported
	}

	if msg.Header == nil || msg.Subscription == nil || msg.Conn == nil {
		return nil, ErrNotReceivedMessage
	}

	if msg.Subscription.AckMode() == AckAuto {
		if ack {
			// not much point sending an ACK to an auto subscription
			return nil, nil
		} else {
			// sending a NACK for an ack:auto subscription makes no
			// sense
			return nil, ErrCannotNackAutoSub
		}
	}

	var f *frame.Frame
	if ack {
		f = frame.New(frame.ACK)
	} else {
		f = frame.New(frame.NACK)
	}

	switch c.version {
	case V10, V11:
		f.Header.Add(frame.Subscription, msg.Subscription.Id())
		if messageId, ok := msg.Header.Contains(frame.MessageId); ok {
			f.Header.Add(frame.MessageId, messageId)
		} else {
			return nil, missingHeader(frame.MessageId)
		}
	case V12:
		if ack, ok := msg.Header.Contains(frame.Ack); ok {
			f.Header.Add(frame.Id, ack)
		} else {
			return nil, missingHeader(frame.Ack)
		}
	}

	return f, nil
}

// Unsubscribes and closes the channel C.
func (s *Subscription) Unsubscribe() error {
	if s.completed {
		return ErrCompletedSubscription
	}
	f := frame.New(frame.UNSUBSCRIBE, frame.Id, s.id)
	s.conn.sendFrame(f)
	s.completed = true
	close(s.C)
	return nil
}

// Commit will commit the transaction. All messages and acknowledgements
// sent to the STOMP server on this transaction will be processed atomically.
func (tx *Transaction) Commit() error {
	if tx.completed {
		return ErrCompletedTransaction
	}

	f := frame.New(frame.COMMIT, frame.Transaction, tx.id)
	tx.conn.sendFrame(f)
	tx.completed = true

	return nil
}

// Sends a RECEIPT frame to the client if the frame f contains
// a receipt header. If the frame does contain a receipt header,
// it will be removed from the frame.
func (c *Conn) sendReceiptImmediately(f *frame.Frame) error {
	if receipt, ok := f.Header.Contains(frame.Receipt); ok {
		// Remove the receipt header from the frame. This is handy
		// for transactions, because the frame has its receipt
		// header removed prior to entering the transaction store.
		// When the frame is processed upon transaction commit, it
		// will not have a receipt header anymore.
		f.Header.Del(frame.Receipt)
		return c.sendImmediately(frame.New(frame.RECEIPT,
			frame.ReceiptId, receipt))
	}
	return nil
}

func (s *TopicSuite) TestTopicWithOneSubscription(c *C) {
	sub := &fakeSubscription{}

	topic := newTopic("destination")
	topic.Subscribe(sub)

	f := frame.New(frame.MESSAGE,
		frame.Destination, "destination")

	topic.Enqueue(f)

	c.Assert(len(sub.Frames), Equals, 1)
	c.Assert(sub.Frames[0], Equals, f)
}

// Send an ERROR frame to the client and immediately. The error
// message is derived from err. If f is non-nil, it is the frame
// whose contents have caused the error. Include the receipt-id
// header if the frame contains a receipt header.
func (c *Conn) sendErrorImmediately(err error, f *frame.Frame) {
	errorFrame := frame.New(frame.ERROR,
		frame.Message, err.Error())

	// Include a receipt-id header if the frame that prompted the error had
	// a receipt header (as suggested by the STOMP protocol spec).
	if f != nil {
		if receipt, ok := f.Header.Contains(frame.Receipt); ok {
			errorFrame.Header.Add(frame.ReceiptId, receipt)
		}
	}

	// send the frame to the client, ignore any error condition
	// because we are about to close the connection anyway
	_ = c.sendImmediately(errorFrame)
}