()
| 41 | } |
| 42 | |
| 43 | func (b *fetchResponseBuilder) bytes() []byte { |
| 44 | if b.rendered == nil { |
| 45 | b.rendered = newWB().call(func(wb *kafkaWriteBuffer) { |
| 46 | wb.writeInt32(b.header.throttle) |
| 47 | wb.writeInt16(b.header.errorCode) |
| 48 | wb.writeInt32(b.header.sessionID) |
| 49 | wb.writeInt32(1) // num topics |
| 50 | wb.writeString(b.header.topic) |
| 51 | wb.writeInt32(1) // how many partitions |
| 52 | wb.writeInt32(b.header.partition) |
| 53 | wb.writeInt16(b.header.partitionErrorCode) |
| 54 | wb.writeInt64(b.header.highWatermarkOffset) |
| 55 | wb.writeInt64(b.header.lastStableOffset) |
| 56 | wb.writeInt64(b.header.logStartOffset) |
| 57 | wb.writeInt32(-1) // num aborted tx |
| 58 | wb.writeBytes(newWB().call(func(wb *kafkaWriteBuffer) { |
| 59 | for _, msgSet := range b.msgSets { |
| 60 | wb.Write(msgSet.bytes()) |
| 61 | } |
| 62 | })) |
| 63 | }) |
| 64 | } |
| 65 | return b.rendered |
| 66 | } |
| 67 | |
| 68 | func (b *fetchResponseBuilder) Len() int { |
| 69 | return len(b.bytes()) |
no test coverage detected