docs: public static Flux<String> create(final Flux<String> source, final String delimiter, final int maxByteArraySize, final long maxMillisecondsBetweenEmits) {
          return Flux.defer(() -> {
              final int delimiterSize = delimiter.getBytes().length;
              final AtomicInteger byteSize = new AtomicInteger(0);
              final AtomicLong lastTime = new AtomicLong(0);
  
              final DirectProcessor<Void> intervalEnd = DirectProcessor.create();
  
              final Flux<String> hearbeat = Flux.interval(Duration.ofMillis(maxMillisecondsBetweenEmits))
                      .map(l -> "")
                      .takeUntilOther(intervalEnd);
  
              // Create a stream that emits at least once every $maxMillisecondsBetweenEmits, to avoid long pauses between
              // buffer flushes when the source doesn't emit for a while.
              final Flux<String> sourceWithEmptyStringKeepAlive = source
                      .doOnTerminate(intervalEnd::onComplete)
                      .mergeWith(hearbeat);
  
              return sourceWithEmptyStringKeepAlive
                      .bufferUntil(line -> {
                          final int bytesLength = line.getBytes().length;
                          final long now = System.currentTimeMillis();
                          // Update last time to now if this is the first time
                          lastTime.compareAndSet(0, now);
                          final long last = lastTime.get();
                          long diff;
                          if (last != 0L) {
                              diff = now - last;
                              if (diff > maxMillisecondsBetweenEmits && byteSize.get() > 0) {
                                  // This creates a buffer, reset size
                                  byteSize.set(bytesLength);
                                  lastTime.compareAndSet(last, now);
                                  return true;
                              }
                          }
  
                          int additionalBytes = bytesLength;
                          if (additionalBytes > 0 && byteSize.get() > 0) {
                              additionalBytes += delimiterSize;  // Make up for the delimiter that's added when joining the strings
                          }
  
                          final int projectedBytes = byteSize.addAndGet(additionalBytes);
  
                          if (projectedBytes > maxByteArraySize) {
                              // This creates a buffer, reset size
                              byteSize.set(bytesLength);
                              lastTime.compareAndSet(last, now);
                              return true;
                          }
  
                          return false;
                      }, true)
                      .map(lines -> {
                          lines.removeIf(String::isEmpty); // Ignore empty messages
                          return String.join(delimiter, lines);
                      });
          });
      }