better 'none' compression handling on produce

client/client.js

@@ -64,6 +64,7 @@ module.exports = function (
 			return cb(err)
 		}
 		if (!written) {
+			logger.info('connection', 'wait')
 			this.ready = false
 		}
 		this.receiver.push(request, cb)

client/produce-request.js

@@ -56,7 +56,8 @@ module.exports = function (
 		if (err) {
 			return cb(err)
 		}
-		if (buffer.length > this.topic.maxMessageSize) {
+		if (this.topic.compression !== Message.compression.NONE &&
+		    buffer.length > this.topic.maxMessageSize) {
 			return cb(new ProduceError("message too big", buffer.length))
 		}
 		var header = new RequestHeader(
@@ -66,11 +67,11 @@ module.exports = function (
 			this.partitionId
 		)
 		try {
-			header.serialize(stream)
+			var written = header.serialize(stream)
 			var mlen = new Buffer(4)
 			mlen.writeUInt32BE(buffer.length, 0)
-			stream.write(mlen)
-			var written = stream.write(buffer)
+			written = stream.write(mlen) && written
+			written = stream.write(buffer) && written
 		}
 		catch (e) {
 			err = e

index.js

@@ -34,7 +34,7 @@ module.exports = function (options) {
 	var Broker = require('./broker')(logger, inherits, EventEmitter, Client)
 	var BrokerPool = require('./broker-pool')(logger, inherits, EventEmitter)
 	var Partition = require('./partition')(logger, inherits, EventEmitter, Broker)
-	var PartitionSet = require('./partition-set')(logger, inherits, EventEmitter, Partition)
+	var PartitionSet = require('./partition-set')(logger, inherits, EventEmitter)
 	var MessageBuffer = require('./message-buffer')(inherits, EventEmitter)
 	var Topic = require('./topic')(logger, inherits, Stream, MessageBuffer, Partition, PartitionSet)
 	var StaticConnector = require('./static-connector')(logger, inherits, EventEmitter, Broker)

partition-set.js

@@ -1,8 +1,7 @@
 module.exports = function (
 	logger,
 	inherits,
-	EventEmitter,
-	Partition) {
+	EventEmitter) {
 
 	function PartitionSet() {
 		this.partitionsByName = {}
@@ -69,11 +68,11 @@ module.exports = function (
 	}
 
 	PartitionSet.prototype.nextWritable = function () {
-		var partition = Partition.nil
+		var partition = null
 		for (var i = 0; i < this.partitions.length; i++) {
 			partition = this.next()
 			if (partition.isWritable() && partition.isReady()) {
-				break;
+				return partition
 			}
 		}
 		return partition
@@ -107,8 +106,6 @@ module.exports = function (
 		this.readable().forEach(stopPartition)
 	}
 
-	PartitionSet.nil = new PartitionSet()
-
 	function readableChanged(partition) {
 		if (partition.isReadable()) {
 			this.readables[partition.name] = partition

topic.js

@@ -166,16 +166,17 @@ module.exports = function (
 		}
 	}
 
-	// Readable Stream
-
 	Topic.prototype.error = function (err) {
 		if (!this.paused) {
 			this.pause()
 		}
 		logger.info('topic', this.name, 'error', err.message)
 		this.emit('error', err)
+		return false
 	}
 
+	// Readable Stream
+
 	Topic.prototype.pause = function () {
 		logger.info('pause', this.name)
 		this.paused = true
@@ -206,6 +207,9 @@ module.exports = function (
 			encoding = encoding || 'utf8'
 			data = new Buffer(data, encoding)
 		}
+		if (data.length > this.maxMessageSize) {
+			return this.error(new Error("message too big"))
+		}
 		return this.produceBuffer.push(data)
 	}