Method to put records in the specified delivery stream by reading
contents from sample input file using PutRecord API.

@throws IOException
546844
protected static void putRecordIntoDeliveryStream() throws IOException {
        try (InputStream inputStream =
                Thread.currentThread().getContextClassLoader().getResourceAsStream(PUT_RECORD_STREAM_SOURCE)) {
            if (inputStream == null) {
                throw new FileNotFoundException("Could not find file " + PUT_RECORD_STREAM_SOURCE);
            }

            try (BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream))) {
                String line = null;
                while ((line = reader.readLine()) != null) {
                    PutRecordRequest putRecordRequest = new PutRecordRequest();
                    putRecordRequest.setDeliveryStreamName(deliveryStreamName);

                    String data = line + "\n";
                    Record record = createRecord(data);
                    putRecordRequest.setRecord(record);

                    // Put record into the DeliveryStream
                    firehoseClient.putRecord(putRecordRequest);
                }
            }
        }
    }
847804
@Override
	public void serializeRecord(T record) throws IOException {
		if (CHECKED) {
			if (dataBuffer.hasRemaining()) {
				throw new IllegalStateException("Pending serialization of previous record.");
			}
		}

		serializationBuffer.clear();
		lengthBuffer.clear();

		// write data and length
		record.write(serializationBuffer);

		int len = serializationBuffer.length();
		lengthBuffer.putInt(0, len);

		dataBuffer = serializationBuffer.wrapAsByteBuffer();
	}
1254060
private void init() throws IOException {
        if (initialized.compareAndSet(false, true)) {
            log.debug("Init: {}", mode);
            // instance an executor for queue handling
            scheduledExecutorService = Executors.newScheduledThreadPool(schedulerThreadSize, new CustomizableThreadFactory("FileConsumerExecutor-"));
            // if the path is null, the consumer has been uninitialized
            if (path != null) {
                if (log.isDebugEnabled()) {
                    Path parent = path.getParent();
                    log.debug("Parent abs: {} dir: {}", parent.isAbsolute(), Files.isDirectory(parent));
                }
                if (IClientStream.MODE_APPEND.equals(mode)) {
                    if (Files.notExists(path)) {
                        throw new IOException("File to be appended doesnt exist, verify the record mode");
                    }
                    log.debug("Path: {}\nRead: {} write: {} size: {}", path, Files.isReadable(path), Files.isWritable(path), Files.size(path));
                    writer = new FLVWriter(path, true);
                } else if (IClientStream.MODE_RECORD.equals(mode)) {
                    try {
                        // delete existing file
                        if (Files.deleteIfExists(path)) {
                            log.debug("File deleted");
                        }
                        // ensure parent dirs exist
                        Files.createDirectories(path.getParent());
                        // create the file
                        path = Files.createFile(path);
                    } catch (IOException ioe) {
                        log.error("File creation error: {}", ioe);
                    }
                    if (!Files.isWritable(path)) {
                        throw new IOException("File is not writable");
                    }
                    log.debug("Path: {}\nRead: {} write: {}", path, Files.isReadable(path), Files.isWritable(path));
                    writer = new FLVWriter(path, false);
                    if (audioConfigurationTag != null) {
                        writer.writeTag(audioConfigurationTag);
                    }
                    if (videoConfigurationTag != null) {
                        writer.writeTag(videoConfigurationTag);
                        gotVideoKeyframe = true;
                    }
                } else {
                    //throw new IllegalStateException(String.format("Illegal mode type: %s", mode));
                    try {
                        // delete existing file since we're not recording nor appending
                        if (Files.deleteIfExists(path)) {
                            log.debug("File deleted");
                        }
                    } catch (IOException ioe) {
                        log.error("File creation error: {}", ioe);
                    }
                }
            } else {
                log.warn("Consumer is uninitialized");
            }
            log.debug("Init - complete");
        }
    }
184860
static byte[] readAllBytes(InputStream stream) throws IOException {
    int bytesRead = 0;
    byte[] currentChunk = new byte[255];
    final List<byte[]> chunks = new ArrayList<byte[]>();
    while ((bytesRead = stream.read(currentChunk)) != -1) {
      chunks.add(copyOf(currentChunk, bytesRead));
    }

    if (chunks.size() <= 0) {
      throw new IOException("Stream did not contain any data. Please provide data to upload.");
    }

    int size = 0;
    for (byte[] chunk : chunks) {
      size += chunk.length;
    }

    final byte[] content = new byte[size];
    int position = 0;
    for (byte[] chunk : chunks) {
      arraycopy(chunk, 0, content, position, chunk.length);
      position += chunk.length;
    }
    return content;
  }
446496
public String put(File file) throws IOException {

        if (!connection.isOpen())
            connection.open();

        connection.prepare("PUT");

        String contentType = connection.getHeader("Content-Type");
        if (contentType == null)
            contentType = connection.getHeader("content-type");
        if (contentType == null || contentType.isEmpty()) {
            /**
             * Default it to application/octet-stream if nothing has been specified
             */
            connection.setHeader("Content-Type", "application/octet-stream");
        }

        OutputStream outStream = connection.getOutputStream();
        InputStream inStream = new FileInputStream(file);

        byte[] buf = new byte[1024];
        int len = 0;
        while (len != -1)
        {
          len = inStream.read(buf);
          if (len > 0)
            outStream.write(buf, 0, len);
        }

        inStream.close();
        outStream.close();

        BufferedReader reader = new BufferedReader(new InputStreamReader(connection.getConnection().getInputStream()));
        StringBuffer responseBuffer = new StringBuffer();
        String line = null;
        while ((line = reader.readLine()) != null) {
            responseBuffer.append(line).append('\n');
        }

        reader.close();

        connection.getConnection().disconnect();

        response = new HttpResponse(responseBuffer.toString().getBytes());
        response.setCode(connection.getConnection().getResponseCode());
        if (response.getCode() < 200 || response.getCode() >= 300)
        {
            response.setMessage(connection.getConnection().getResponseMessage());
            throw new IOException("Error uploading file: " + response.getCode() + " -- " + response.getMessage());
        }

        return getResponse();
    }
535675
private void updateInputStream() throws IOException {
    if (mInputStream != null && (mInputStream.mFile.isIncompleteLog() || !mInputStream.isDone())) {
      return;
    }

    if (mInputStream != null) {
      mInputStream.close();
      mInputStream = null;
    }
    if (mFilesToProcess.isEmpty()) {
      UfsJournalSnapshot snapshot = UfsJournalSnapshot.getSnapshot(mJournal);
      if (snapshot.getCheckpoints().isEmpty() && snapshot.getLogs().isEmpty()) {
        return;
      }

      int index = 0;
      if (!snapshot.getCheckpoints().isEmpty()) {
        UfsJournalFile checkpoint = snapshot.getLatestCheckpoint();
        if (mNextSequenceNumber < checkpoint.getEnd()) {
          String location = checkpoint.getLocation().toString();
          LOG.info("Reading checkpoint {}", location);
          mCheckpointStream = new CheckpointInputStream(
              mUfs.open(location, OpenOptions.defaults().setRecoverFailedOpen(true)));
          mNextSequenceNumber = checkpoint.getEnd();
        }
        for (; index < snapshot.getLogs().size(); index++) {
          UfsJournalFile file = snapshot.getLogs().get(index);
          if (file.getEnd() > checkpoint.getEnd()) {
            break;
          }
        }
        // index now points to the first log with mEnd > checkpoint.mEnd.
      }
      for (; index < snapshot.getLogs().size(); index++) {
        UfsJournalFile file = snapshot.getLogs().get(index);
        if ((!mReadIncompleteLog && file.isIncompleteLog())
            || mNextSequenceNumber >= file.getEnd()) {
          continue;
        }
        mFilesToProcess.add(snapshot.getLogs().get(index));
      }
    }

    if (!mFilesToProcess.isEmpty()) {
      mInputStream = new JournalInputStream(mFilesToProcess.poll());
    }
  }
853886
public E read() throws IOException {
		boolean status = false;
		try {

			// Read header if not already read
			readHeader();

			// Read the records
			E element;
			try {
				do {
					element = readRecord(this.nextExpectedRecordIndex);
					if (!postRecordReadingStage(element)) {
						element = null;
					}
					++this.nextExpectedRecordIndex;
				}
				while (element == null);
			} catch (EOFException e) {
				element = null;
				close();
				postReadingStage(true);
			}
			status = true;

			return element;
		} finally {
			if (this.taskProgression != null) {
				this.taskProgression.setValue(this.buffer.position() + HEADER_BYTES);
			}
			if (!status) {
				close();
				postReadingStage(status);
			}
		}
	}
777217
@Override
    public synchronized void writeRecord(String record) {
        long length = record.length() + LoggingConstants.nlen;
        PrintStream ps = getPrintStream(length);
        ps.println(record);
        if (ps.checkError()) {
            setStreamStatus(StreamStatus.CLOSED, null, null, DummyOutputStream.psInstance);
            // to avoid junit test to print an error message
            if (System.getProperty("test.classesDir") == null && System.getProperty("test.buildDir") == null) {
                System.err.println(Tr.formatMessage(getTc(), "FAILED_TO_WRITE_LOG", new Object[] { getPrimaryFile().getAbsolutePath() }));
            }
        }
    }
730730
private void initStream() throws IOException {

        if (m_out == null) {
            if (!m_writeOnlyToBuffer) {
                // we can use the parents output stream
                if (m_cachingRequired || (m_controller.getResponseStackSize() > 1)) {
                    // we are allowed to cache our results (probably to construct a new cache entry)
                    m_out = new CmsFlexResponse.CmsServletOutputStream(m_res.getOutputStream());
                } else {
                    // we are not allowed to cache so we just use the parents output stream
                    m_out = (CmsFlexResponse.CmsServletOutputStream)m_res.getOutputStream();
                }
            } else {
                // construct a "buffer only" output stream
                m_out = new CmsFlexResponse.CmsServletOutputStream();
            }
        }
        if (m_writer == null) {
            // create a PrintWriter that uses the encoding required for the request context
            m_writer = new PrintWriter(new BufferedWriter(new OutputStreamWriter(m_out, m_encoding)), false);
        }
    }
742959
private void writeFileCreationRecord() throws IOException
   {
      ProjectProperties properties = m_projectFile.getProjectProperties();

      m_buffer.setLength(0);
      m_buffer.append("MPX");
      m_buffer.append(m_delimiter);
      m_buffer.append(properties.getMpxProgramName());
      m_buffer.append(m_delimiter);
      m_buffer.append(properties.getMpxFileVersion());
      m_buffer.append(m_delimiter);
      m_buffer.append(properties.getMpxCodePage());
      m_buffer.append(MPXConstants.EOL);
      m_writer.write(m_buffer.toString());
   }
379098
public int putRawRecordData(Rec record)
    {
        if (this.getNodeType() == BaseMessageRecordDesc.NON_UNIQUE_NODE)
            this.getMessage().createNewNode(this.getFullKey(null));
        return super.putRawRecordData(record);
    }
242887
private void putObject(WebContext ctx, Bucket bucket, String id, InputStreamHandler inputStream)
            throws IOException {
        StoredObject object = bucket.getObject(id);
        if (inputStream == null) {
            signalObjectError(ctx, HttpResponseStatus.BAD_REQUEST, "No content posted");
            return;
        }
        try (FileOutputStream out = new FileOutputStream(object.getFile())) {
            ByteStreams.copy(inputStream, out);
        }

        Map<String, String> properties = Maps.newTreeMap();
        for (String name : ctx.getRequest().headers().names()) {
            String nameLower = name.toLowerCase();
            if (nameLower.startsWith("x-amz-meta-") || "content-md5".equals(nameLower) || "content-type".equals(
                    nameLower) || "x-amz-acl".equals(nameLower)) {
                properties.put(name, ctx.getHeader(name));
            }
        }

        HashCode hash = Files.hash(object.getFile(), Hashing.md5());
        String md5 = BaseEncoding.base64().encode(hash.asBytes());
        String contentMd5 = properties.get("Content-MD5");
        if (properties.containsKey("Content-MD5") && !md5.equals(contentMd5)) {
            object.delete();
            signalObjectError(ctx,
                              HttpResponseStatus.BAD_REQUEST,
                              Strings.apply("Invalid MD5 checksum (Input: %s, Expected: %s)", contentMd5, md5));
            return;
        }

        String etag = BaseEncoding.base16().encode(hash.asBytes());
        properties.put(HTTP_HEADER_NAME_ETAG, etag);
        object.storeProperties(properties);
        Response response = ctx.respondWith();
        response.addHeader(HTTP_HEADER_NAME_ETAG, etag(etag)).status(HttpResponseStatus.OK);
        response.addHeader(HttpHeaderNames.ACCESS_CONTROL_EXPOSE_HEADERS, HTTP_HEADER_NAME_ETAG);
        signalObjectSuccess(ctx);
    }
645823
protected void gotoEOR(ArchiveRecord record) throws IOException {
        if (getIn().available() <= 0) {
            return;
        }
        
        // Remove any trailing LINE_SEPARATOR
        int c = -1;
        while (getIn().available() > 0) {
            if (getIn().markSupported()) {
                getIn().mark(1);
            }
            c = getIn().read();
            if (c != -1) {
                if (c == LINE_SEPARATOR) {
                    continue;
                }
                if (getIn().markSupported()) {
                    // We've overread.  We're probably in next record.  There is
                    // no way of telling for sure. It may be dross at end of
                    // current record. Backup.
                        getIn().reset();
                    break;
                }
                ArchiveRecordHeader h = (getCurrentRecord() != null)?
                    record.getHeader(): null;
                throw new IOException("Read " + (char)c +
                    " when only " + LINE_SEPARATOR + " expected. " + 
                    getReaderIdentifier() + ((h != null)?
                        h.getHeaderFields().toString(): ""));
            }
        }
    }
1207470
public byte[] readRecord() throws IOException {
        if (this.debug) {
            System.err.println("ReadRecord: recIdx = " + this.currRecIdx + " blkIdx = " + this.currBlkIdx);
        }

        if (this.inStream == null) {
            throw new IOException("reading from an output buffer");
        }

        if (this.currRecIdx >= this.recsPerBlock) {
            if (!this.readBlock()) {
                return null;
            }
        }

        byte[] result = new byte[this.recordSize];

        System.arraycopy(this.blockBuffer, (this.currRecIdx * this.recordSize), result, 0, this.recordSize);

        this.currRecIdx++;

        return result;
    }
785609
@Override
	public void insert(T record) throws IOException {
		if (closed) {
			return;
		}

		final int hashCode = MathUtils.jenkinsHash(buildSideComparator.hash(record));
		final int bucket = hashCode & numBucketsMask;
		final int bucketSegmentIndex = bucket >>> numBucketsPerSegmentBits; // which segment contains the bucket
		final MemorySegment bucketSegment = bucketSegments[bucketSegmentIndex];
		final int bucketOffset = (bucket & numBucketsPerSegmentMask) << bucketSizeBits; // offset of the bucket in the segment
		final long firstPointer = bucketSegment.getLong(bucketOffset);

		try {
			final long newFirstPointer = recordArea.appendPointerAndRecord(firstPointer, record);
			bucketSegment.putLong(bucketOffset, newFirstPointer);
		} catch (EOFException ex) {
			compactOrThrow();
			insert(record);
			return;
		}

		numElements++;
		resizeTableIfNecessary();
	}
148290
protected void setReadingPosition(int recordIndex, int byteIndex) throws IOException {
		if (this.seekEnabled) {
			this.nextExpectedRecordIndex = recordIndex;
			this.buffer.position(byteIndex);
		} else {
			throw new SeekOperationDisabledException();
		}
	}
763057
@Override
  public void write(byte[] record) throws IOException {
    Preconditions.checkNotNull(record);

    byte[] toWrite = record;
    if (this.recordDelimiter.isPresent()) {
      toWrite = Arrays.copyOf(record, record.length + 1);
      toWrite[toWrite.length - 1] = this.recordDelimiter.get();
    }
    if (this.prependSize) {
      long recordSize = toWrite.length;
      ByteBuffer buf = ByteBuffer.allocate(Longs.BYTES);
      buf.putLong(recordSize);
      toWrite = ArrayUtils.addAll(buf.array(), toWrite);
    }
    this.stagingFileOutputStream.write(toWrite);
    this.bytesWritten += toWrite.length;
    this.recordsWritten++;
  }