Rename bbstreamer to astreamer.

I (rhaas) intended "bbstreamer" to stand for "base backup streamer,"
but that implies that this infrastructure can only ever be used by
pg_basebackup.  In fact, it is a generally useful way of streaming
data from a tar or compressed tar file, and it could be extended to
work with other archive formats as well if we ever wanted to do that.
Hence, rename it to "astreamer" (archive streamer) in preparation for
reusing the infrastructure from pg_verifybackup (and perhaps
eventually also other utilities, such as pg_combinebackup or
pg_waldump).

This is purely a renaming commit. Comment adjustments and relocation
of the actual code to someplace from which it can be reused are left
to future commits.

Amul Sul, reviewed by Sravan Kumar and by me.

Discussion: http://postgr.es/m/CAAJ_b94StvLWrc_p4q-f7n3OPfr6GhL8_XuAg2aAaYZp1tF-nw@mail.gmail.com

1 files changed, 52 insertions, 52 deletions

diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c
index 8f3dd04fd22..1966ada69c9 100644
--- a/src/bin/pg_basebackup/pg_basebackup.c
+++ b/src/bin/pg_basebackup/pg_basebackup.c
@@ -26,8 +26,8 @@
 #endif
 
 #include "access/xlog_internal.h"
+#include "astreamer.h"
 #include "backup/basebackup.h"
-#include "bbstreamer.h"
 #include "common/compression.h"
 #include "common/file_perm.h"
 #include "common/file_utils.h"
@@ -57,8 +57,8 @@ typedef struct ArchiveStreamState
 {
 	int			tablespacenum;
 	pg_compress_specification *compress;
-	bbstreamer *streamer;
-	bbstreamer *manifest_inject_streamer;
+	astreamer  *streamer;
+	astreamer  *manifest_inject_streamer;
 	PQExpBuffer manifest_buffer;
 	char		manifest_filename[MAXPGPATH];
 	FILE	   *manifest_file;
@@ -67,7 +67,7 @@ typedef struct ArchiveStreamState
 typedef struct WriteTarState
 {
 	int			tablespacenum;
-	bbstreamer *streamer;
+	astreamer  *streamer;
 } WriteTarState;
 
 typedef struct WriteManifestState
@@ -199,11 +199,11 @@ static void verify_dir_is_empty_or_create(char *dirname, bool *created, bool *fo
 static void progress_update_filename(const char *filename);
 static void progress_report(int tablespacenum, bool force, bool finished);
 
-static bbstreamer *CreateBackupStreamer(char *archive_name, char *spclocation,
-										bbstreamer **manifest_inject_streamer_p,
-										bool is_recovery_guc_supported,
-										bool expect_unterminated_tarfile,
-										pg_compress_specification *compress);
+static astreamer *CreateBackupStreamer(char *archive_name, char *spclocation,
+									   astreamer **manifest_inject_streamer_p,
+									   bool is_recovery_guc_supported,
+									   bool expect_unterminated_tarfile,
+									   pg_compress_specification *compress);
 static void ReceiveArchiveStreamChunk(size_t r, char *copybuf,
 									  void *callback_data);
 static char GetCopyDataByte(size_t r, char *copybuf, size_t *cursor);
@@ -1053,19 +1053,19 @@ ReceiveCopyData(PGconn *conn, WriteDataCallback callback,
  * the options selected by the user.  We may just write the results directly
  * to a file, or we might compress first, or we might extract the tar file
  * and write each member separately. This function doesn't do any of that
- * directly, but it works out what kind of bbstreamer we need to create so
+ * directly, but it works out what kind of astreamer we need to create so
  * that the right stuff happens when, down the road, we actually receive
  * the data.
  */
-static bbstreamer *
+static astreamer *
 CreateBackupStreamer(char *archive_name, char *spclocation,
-					 bbstreamer **manifest_inject_streamer_p,
+					 astreamer **manifest_inject_streamer_p,
 					 bool is_recovery_guc_supported,
 					 bool expect_unterminated_tarfile,
 					 pg_compress_specification *compress)
 {
-	bbstreamer *streamer = NULL;
-	bbstreamer *manifest_inject_streamer = NULL;
+	astreamer  *streamer = NULL;
+	astreamer  *manifest_inject_streamer = NULL;
 	bool		inject_manifest;
 	bool		is_tar,
 				is_tar_gz,
@@ -1160,9 +1160,9 @@ CreateBackupStreamer(char *archive_name, char *spclocation,
 			directory = psprintf("%s/%s", basedir, spclocation);
 		else
 			directory = get_tablespace_mapping(spclocation);
-		streamer = bbstreamer_extractor_new(directory,
-											get_tablespace_mapping,
-											progress_update_filename);
+		streamer = astreamer_extractor_new(directory,
+										   get_tablespace_mapping,
+										   progress_update_filename);
 	}
 	else
 	{
@@ -1188,27 +1188,27 @@ CreateBackupStreamer(char *archive_name, char *spclocation,
 		}
 
 		if (compress->algorithm == PG_COMPRESSION_NONE)
-			streamer = bbstreamer_plain_writer_new(archive_filename,
-												   archive_file);
+			streamer = astreamer_plain_writer_new(archive_filename,
+												  archive_file);
 		else if (compress->algorithm == PG_COMPRESSION_GZIP)
 		{
 			strlcat(archive_filename, ".gz", sizeof(archive_filename));
-			streamer = bbstreamer_gzip_writer_new(archive_filename,
-												  archive_file, compress);
+			streamer = astreamer_gzip_writer_new(archive_filename,
+												 archive_file, compress);
 		}
 		else if (compress->algorithm == PG_COMPRESSION_LZ4)
 		{
 			strlcat(archive_filename, ".lz4", sizeof(archive_filename));
-			streamer = bbstreamer_plain_writer_new(archive_filename,
-												   archive_file);
-			streamer = bbstreamer_lz4_compressor_new(streamer, compress);
+			streamer = astreamer_plain_writer_new(archive_filename,
+												  archive_file);
+			streamer = astreamer_lz4_compressor_new(streamer, compress);
 		}
 		else if (compress->algorithm == PG_COMPRESSION_ZSTD)
 		{
 			strlcat(archive_filename, ".zst", sizeof(archive_filename));
-			streamer = bbstreamer_plain_writer_new(archive_filename,
-												   archive_file);
-			streamer = bbstreamer_zstd_compressor_new(streamer, compress);
+			streamer = astreamer_plain_writer_new(archive_filename,
+												  archive_file);
+			streamer = astreamer_zstd_compressor_new(streamer, compress);
 		}
 		else
 		{
@@ -1222,7 +1222,7 @@ CreateBackupStreamer(char *archive_name, char *spclocation,
 		 * into it.
 		 */
 		if (must_parse_archive)
-			streamer = bbstreamer_tar_archiver_new(streamer);
+			streamer = astreamer_tar_archiver_new(streamer);
 		progress_update_filename(archive_filename);
 	}
 
@@ -1241,9 +1241,9 @@ CreateBackupStreamer(char *archive_name, char *spclocation,
 	if (spclocation == NULL && writerecoveryconf)
 	{
 		Assert(must_parse_archive);
-		streamer = bbstreamer_recovery_injector_new(streamer,
-													is_recovery_guc_supported,
-													recoveryconfcontents);
+		streamer = astreamer_recovery_injector_new(streamer,
+												   is_recovery_guc_supported,
+												   recoveryconfcontents);
 	}
 
 	/*
@@ -1253,9 +1253,9 @@ CreateBackupStreamer(char *archive_name, char *spclocation,
 	 * we're talking to such a server we'll need to add the terminator here.
 	 */
 	if (must_parse_archive)
-		streamer = bbstreamer_tar_parser_new(streamer);
+		streamer = astreamer_tar_parser_new(streamer);
 	else if (expect_unterminated_tarfile)
-		streamer = bbstreamer_tar_terminator_new(streamer);
+		streamer = astreamer_tar_terminator_new(streamer);
 
 	/*
 	 * If the user has requested a server compressed archive along with
@@ -1264,11 +1264,11 @@ CreateBackupStreamer(char *archive_name, char *spclocation,
 	if (format == 'p')
 	{
 		if (is_tar_gz)
-			streamer = bbstreamer_gzip_decompressor_new(streamer);
+			streamer = astreamer_gzip_decompressor_new(streamer);
 		else if (is_tar_lz4)
-			streamer = bbstreamer_lz4_decompressor_new(streamer);
+			streamer = astreamer_lz4_decompressor_new(streamer);
 		else if (is_tar_zstd)
-			streamer = bbstreamer_zstd_decompressor_new(streamer);
+			streamer = astreamer_zstd_decompressor_new(streamer);
 	}
 
 	/* Return the results. */
@@ -1307,10 +1307,10 @@ ReceiveArchiveStream(PGconn *conn, pg_compress_specification *compress)
 	if (state.manifest_inject_streamer != NULL &&
 		state.manifest_buffer != NULL)
 	{
-		bbstreamer_inject_file(state.manifest_inject_streamer,
-							   "backup_manifest",
-							   state.manifest_buffer->data,
-							   state.manifest_buffer->len);
+		astreamer_inject_file(state.manifest_inject_streamer,
+							  "backup_manifest",
+							  state.manifest_buffer->data,
+							  state.manifest_buffer->len);
 		destroyPQExpBuffer(state.manifest_buffer);
 		state.manifest_buffer = NULL;
 	}
@@ -1318,8 +1318,8 @@ ReceiveArchiveStream(PGconn *conn, pg_compress_specification *compress)
 	/* If there's still an archive in progress, end processing. */
 	if (state.streamer != NULL)
 	{
-		bbstreamer_finalize(state.streamer);
-		bbstreamer_free(state.streamer);
+		astreamer_finalize(state.streamer);
+		astreamer_free(state.streamer);
 		state.streamer = NULL;
 	}
 }
@@ -1383,8 +1383,8 @@ ReceiveArchiveStreamChunk(size_t r, char *copybuf, void *callback_data)
 				/* End processing of any prior archive. */
 				if (state->streamer != NULL)
 				{
-					bbstreamer_finalize(state->streamer);
-					bbstreamer_free(state->streamer);
+					astreamer_finalize(state->streamer);
+					astreamer_free(state->streamer);
 					state->streamer = NULL;
 				}
 
@@ -1437,8 +1437,8 @@ ReceiveArchiveStreamChunk(size_t r, char *copybuf, void *callback_data)
 				else if (state->streamer != NULL)
 				{
 					/* Archive data. */
-					bbstreamer_content(state->streamer, NULL, copybuf + 1,
-									   r - 1, BBSTREAMER_UNKNOWN);
+					astreamer_content(state->streamer, NULL, copybuf + 1,
+									  r - 1, ASTREAMER_UNKNOWN);
 				}
 				else
 					pg_fatal("unexpected payload data");
@@ -1600,7 +1600,7 @@ ReceiveTarFile(PGconn *conn, char *archive_name, char *spclocation,
 			   bool tablespacenum, pg_compress_specification *compress)
 {
 	WriteTarState state;
-	bbstreamer *manifest_inject_streamer;
+	astreamer  *manifest_inject_streamer;
 	bool		is_recovery_guc_supported;
 	bool		expect_unterminated_tarfile;
 
@@ -1636,16 +1636,16 @@ ReceiveTarFile(PGconn *conn, char *archive_name, char *spclocation,
 			pg_fatal("out of memory");
 
 		/* Inject it into the output tarfile. */
-		bbstreamer_inject_file(manifest_inject_streamer, "backup_manifest",
-							   buf.data, buf.len);
+		astreamer_inject_file(manifest_inject_streamer, "backup_manifest",
+							  buf.data, buf.len);
 
 		/* Free memory. */
 		termPQExpBuffer(&buf);
 	}
 
 	/* Cleanup. */
-	bbstreamer_finalize(state.streamer);
-	bbstreamer_free(state.streamer);
+	astreamer_finalize(state.streamer);
+	astreamer_free(state.streamer);
 
 	progress_report(tablespacenum, true, false);
 
@@ -1663,7 +1663,7 @@ ReceiveTarCopyChunk(size_t r, char *copybuf, void *callback_data)
 {
 	WriteTarState *state = callback_data;
 
-	bbstreamer_content(state->streamer, NULL, copybuf, r, BBSTREAMER_UNKNOWN);
+	astreamer_content(state->streamer, NULL, copybuf, r, ASTREAMER_UNKNOWN);
 
 	totaldone += r;
 	progress_report(state->tablespacenum, false, false);