external
/
postgres-decoderbufs
mirror of https://github.com/debezium/postgres-decoderbufs.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Syncing up code so others can see latest

pull/1/head
Xavier Stevens 2015-02-10 10:30:39 -08:00
parent becb332b8c
commit 744b29fa7d
1 changed files with 47 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
src/decoderbufs.c
View File

@ -106,6 +106,8 @@ void _PG_output_plugin_init(OutputPluginCallbacks *cb) {
/* initialize this plugin */ /* initialize this plugin */
static void pg_decode_startup(LogicalDecodingContext *ctx, static void pg_decode_startup(LogicalDecodingContext *ctx,
OutputPluginOptions *opt, bool is_init) { OutputPluginOptions *opt, bool is_init) {
elog(INFO, "Entering startup callback");
ListCell *option; ListCell *option;
DecoderData *data; DecoderData *data;
@ -114,7 +116,6 @@ static void pg_decode_startup(LogicalDecodingContext *ctx,
ctx->context, "decoderbufs context", ALLOCSET_DEFAULT_MINSIZE, ctx->context, "decoderbufs context", ALLOCSET_DEFAULT_MINSIZE,
ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE); ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE);
data->debug_mode = false; data->debug_mode = false;
ctx->output_plugin_private = data;
opt->output_type = OUTPUT_PLUGIN_BINARY_OUTPUT; opt->output_type = OUTPUT_PLUGIN_BINARY_OUTPUT;
foreach(option, ctx->output_plugin_options) { foreach(option, ctx->output_plugin_options) {
@ -142,15 +143,18 @@ static void pg_decode_startup(LogicalDecodingContext *ctx,
errmsg("option \"%s\" = \"%s\" is unknown", elem->defname, errmsg("option \"%s\" = \"%s\" is unknown", elem->defname,
elem->arg ? strVal(elem->arg) : "(null)"))); elem->arg ? strVal(elem->arg) : "(null)")));
} }
}
// set PostGIS geometry type id (these are dynamic unfortunately) // set PostGIS geometry type id (these are dynamic unfortunately)
char *geom_oid_str = NULL; char *geom_oid_str = NULL;
char *geog_oid_str = NULL; char *geog_oid_str = NULL;
elog(INFO, "SPI_connect() ... ");
if (SPI_connect() == SPI_ERROR_CONNECT) { if (SPI_connect() == SPI_ERROR_CONNECT) {
elog(NOTICE, "Could not connect to SPI manager to scan for PostGIS types"); elog(WARNING, "Could not connect to SPI manager to scan for PostGIS types");
SPI_finish(); SPI_finish();
return; return;
} }
elog(INFO, "SPI_execute(\"SELECT oid FROM pg_type WHERE typname = 'geometry'\") ... ");
if (SPI_OK_SELECT == if (SPI_OK_SELECT ==
SPI_execute("SELECT oid FROM pg_type WHERE typname = 'geometry'", SPI_execute("SELECT oid FROM pg_type WHERE typname = 'geometry'",
true, 1) && true, 1) &&
@ -161,7 +165,10 @@ static void pg_decode_startup(LogicalDecodingContext *ctx,
elog(NOTICE, "Decoderbufs detected PostGIS geometry type with oid: %s", geom_oid_str); elog(NOTICE, "Decoderbufs detected PostGIS geometry type with oid: %s", geom_oid_str);
geometry_oid = atoi(geom_oid_str); geometry_oid = atoi(geom_oid_str);
} }
} else {
elog(WARNING, "No type oid detected for PostGIS geometry");
} }
elog(INFO, "SPI_execute(\"SELECT oid FROM pg_type WHERE typname = 'geography'\") ... ");
if (SPI_OK_SELECT == if (SPI_OK_SELECT ==
SPI_execute("SELECT oid FROM pg_type WHERE typname = 'geography'", SPI_execute("SELECT oid FROM pg_type WHERE typname = 'geography'",
true, 1) && true, 1) &&
@ -172,9 +179,14 @@ static void pg_decode_startup(LogicalDecodingContext *ctx,
elog(NOTICE, "Decoderbufs detected PostGIS geography type with oid: %s", geog_oid_str); elog(NOTICE, "Decoderbufs detected PostGIS geography type with oid: %s", geog_oid_str);
geography_oid = atoi(geog_oid_str); geography_oid = atoi(geog_oid_str);
} }
} else {
elog(WARNING, "No type oid detected for PostGIS geography");
} }
SPI_finish(); SPI_finish();
}
ctx->output_plugin_private = data;
elog(INFO, "Exiting startup callback");
} }
/* cleanup this plugin's resources */ /* cleanup this plugin's resources */
@ -427,6 +439,8 @@ static void set_datum_value(Decoderbufs__DatumMessage *datum_msg, Oid typid,
datum_msg->datum_point = palloc(sizeof(Decoderbufs__Point)); datum_msg->datum_point = palloc(sizeof(Decoderbufs__Point));
geography_point_as_decoderbufs_point(datum, datum_msg->datum_point); geography_point_as_decoderbufs_point(datum, datum_msg->datum_point);
} else { } else {
elog(DEBUG1, "Encountered unknown typid: %d", typid);
elog(DEBUG1, "PostGIS Geometry OID[%d], Geography OID[%d]", geometry_oid, geography_oid);
output = OidOutputFunctionCall(typoutput, datum); output = OidOutputFunctionCall(typoutput, datum);
int len = strlen(output); int len = strlen(output);
size = sizeof(char) * len; size = sizeof(char) * len;