class PublisherLease implements Closeable {

    void publish(final FlowFile flowFile, final InputStream flowFileContent, final byte[] messageKey, final byte[] demarcatorBytes, final String topic) throws IOException {
        if (tracker == null) {
            tracker = new InFlightMessageTracker(logger);
        }

        try {
            byte[] messageContent;
            if (demarcatorBytes == null || demarcatorBytes.length == 0) {
                if (flowFile.getSize() > maxMessageSize) {
                    tracker.fail(flowFile, new TokenTooLargeException("A message in the stream exceeds the maximum allowed message size of " + maxMessageSize + " bytes."));
                    return;
                }
                // Send FlowFile content as it is, to support sending 0 byte message.
                messageContent = new byte[(int) flowFile.getSize()];
                StreamUtils.fillBuffer(flowFileContent, messageContent);
                publish(flowFile, messageKey, messageContent, topic, tracker);
                return;
            }

            try (final StreamDemarcator demarcator = new StreamDemarcator(flowFileContent, demarcatorBytes, maxMessageSize)) {
                while ((messageContent = demarcator.nextToken()) != null) {
                    publish(flowFile, messageKey, messageContent, topic, tracker);

                    if (tracker.isFailed(flowFile)) {
                        // If we have a failure, don't try to send anything else.
                        return;
                    }
                    tracker.trackEmpty(flowFile);
                }
            } catch (final TokenTooLargeException ttle) {
                tracker.fail(flowFile, ttle);
            }
        } catch (final Exception e) {
            tracker.fail(flowFile, e);
            poison();
            throw e;
        }
    }

    public  PublisherLease(final Producer<byte[], byte[]> producer, final int maxMessageSize, final long maxAckWaitMillis, final ComponentLog logger,
        final boolean useTransactions, final Pattern attributeNameRegex, final Charset headerCharacterSet);

}