sending/receiving large AMQP headers

Bug53Spec.scala

package com.github.sstone.amqp

import akka.testkit.TestProbe
import com.github.sstone.amqp.Amqp._
import com.rabbitmq.client.AMQP.BasicProperties
import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner
import scala.collection.JavaConversions._
import scala.concurrent.duration._

@RunWith(classOf[JUnitRunner])
class Bug53Spec extends ChannelSpec {
  "header values larger then 1024 bytes should be sent/received properly" in {
    val probe = TestProbe()
    val queue = randomQueue
    val consumer = ConnectionOwner.createChildActor(conn, Consumer.props(listener = probe.ref, channelParams = None, autoack = true))
    val producer = ConnectionOwner.createChildActor(conn, ChannelOwner.props())
    Amqp.waitForConnection(system, consumer, producer)
    consumer ! AddStatusListener(probe.ref)
    probe.expectMsg(ChannelOwner.Connected)
    consumer ! AddQueue(queue)
    val Amqp.Ok(AddQueue(_), _) = receiveOne(1 second)

    val body = "test".getBytes("UTF-8")
    val blob = new Array[Byte](30*1024)
    random.nextBytes(blob)
    val props = new BasicProperties.Builder().headers(Map("blob" -> blob)).build()
    producer ! Publish("", queue.name, body, Some(props))
    val Delivery(_, _, properties, body1) = probe.receiveOne(1 second)
    val headers = properties.getHeaders
    assert(body1 === body)
    assert(headers("blob") === blob)
  }
}