 			self.queues[queue].unsubscribe socket.ctag
 	self.subscribe = (queue,socket,listener) ->
 		if not self.queues[queue]
-			self.connection?.queue queue, (Queue) ->
+			self.connection?.queue queue, { arguments: { "x-message-ttl" : 60000 } }, (Queue) ->
 				self.queues[queue] = Queue
 				(Queue.subscribe { ack: false, prefetchCount: 1 }, (message, headers, deliveryInfo) ->
 					listener(message.data.toString())

	self.delete = (queue) ->
		# deletes a queue, unbinding it in the proces
		if not self.queues[queue]
			self.connection?.queue queue, (Queue) ->
				Queue.destroy()
		else
			self.queues[queue]?.destroy()

		self.connection?.exchange exchange, { durable: false, type: 'topic', autoDelete: true, closeChannelOnUnsubscribe: true },  (Exchange) ->
			self.exchanges[exchange] = Exchange
		if queue
			self.connection?.queue queue, { arguments: { "x-message-ttl" : 60000 } }, (Queue) ->
				self.queues[queue] = Queue
				Queue.bind exchange, key
	self.read = (queue,fun) ->
		# reads a message from the given queue and returns the result to the supplied callback
		if not self.queues[queue]
			self.connection?.queue queue, { arguments: { "x-message-ttl" : 60000 } }, (Queue) ->
				self.queues[queue] = Queue
				Queue.get({ noack: true }, fun)
		else
