func (r *receiver) Ordered(msg *cb.Envelope) (messageBatches [][]*cb.Envelope, pending bool) {
ordererConfig, ok := r.sharedConfigFetcher.OrdererConfig()
batchSize := ordererConfig.BatchSize()messageSizeBytes := messageSizeBytes(msg)
if messageSizeBytes > batchSize.PreferredMaxBytes {
// cut pending batch, if it has any messages
if len(r.pendingBatch) > 0 {
messageBatch := r.Cut()
messageBatches = append(messageBatches, messageBatch)
}
// create new batch with single message
messageBatches = append(messageBatches, []*cb.Envelope{msg})
// Record that this batch took no time to fill
r.Metrics.BlockFillDuration.With("channel", r.ChannelID).Observe(0)
return
}
messageWillOverflowBatchSizeBytes := r.pendingBatchSizeBytes+messageSizeBytes > batchSize.PreferredMaxBytesif messageWillOverflowBatchSizeBytes {messageBatch := r.Cut()
r.PendingBatchStartTime = time.Now()
messageBatches = append(messageBatches, messageBatch)
}
r.pendingBatch = append(r.pendingBatch, msg)
r.pendingBatchSizeBytes += messageSizeBytes
pending = trueif uint32(len(r.pendingBatch)) >= batchSize.MaxMessageCount {
logger.Debugf("Batch size met, cutting batch")
messageBatch := r.Cut()
messageBatches = append(messageBatches, messageBatch)
pending = false
}return
}
Orderer: &OrdererDefaults# Orderer Type: The orderer implementation to start
# Available types are "solo" and "kafka"
OrdererType: soloAddresses:
- orderer.example.com:7050# Batch Timeout: The amount of time to wait before creating a batch
BatchTimeout: 2s# Batch Size: Controls the number of messages batched into a block
BatchSize:# Max Message Count: The maximum number of messages to permit in a batch
MaxMessageCount: 10# Absolute Max Bytes: The absolute maximum number of bytes allowed for
# the serialized messages in a batch.
AbsoluteMaxBytes: 99 MB# Preferred Max Bytes: The preferred maximum number of bytes allowed for
# the serialized messages in a batch. A message larger than the preferred
# max bytes will result in a batch larger than preferred max bytes.
PreferredMaxBytes: 512 KB
switch {
case timer != nil && !pending:
// Timer is already running but there are no messages pending, stop the timer
timer = nil
case timer == nil && pending:
// Timer is not already running and there are messages pending, so start it
timer = time.After(ch.support.SharedConfig().BatchTimeout())
logger.Debugf("Just began %s batch timer", ch.support.SharedConfig().BatchTimeout().String())
default:
// Do nothing when:
// 1. Timer is already running and there are messages pending
// 2. Timer is not set and there are no messages pending
}
