Fixes #63 event property configuration is ignored

spi/config/configuration.go

@@ -239,23 +239,23 @@ type IncludedTablesConfig struct {
 }
 
 type TimescaleEventsConfig struct {
-	Read          bool `toml:"read" yaml:"read"`
-	Insert        bool `toml:"insert" yaml:"insert"`
-	Update        bool `toml:"update" yaml:"update"`
-	Delete        bool `toml:"delete" yaml:"delete"`
-	Truncate      bool `toml:"truncate" yaml:"truncate"`
-	Message       bool `toml:"message" yaml:"message"`
-	Compression   bool `toml:"compression" yaml:"compression"`
-	Decompression bool `toml:"decompression" yaml:"decompression"`
+	Read          *bool `toml:"read" yaml:"read"`
+	Insert        *bool `toml:"insert" yaml:"insert"`
+	Update        *bool `toml:"update" yaml:"update"`
+	Delete        *bool `toml:"delete" yaml:"delete"`
+	Truncate      *bool `toml:"truncate" yaml:"truncate"`
+	Message       *bool `toml:"message" yaml:"message"` // deprecated
+	Compression   *bool `toml:"compression" yaml:"compression"`
+	Decompression *bool `toml:"decompression" yaml:"decompression"`
 }
 
 type PostgresqlEventsConfig struct {
-	Read     bool `toml:"read" yaml:"read"`
-	Insert   bool `toml:"insert" yaml:"insert"`
-	Update   bool `toml:"update" yaml:"update"`
-	Delete   bool `toml:"delete" yaml:"delete"`
-	Truncate bool `toml:"truncate" yaml:"truncate"`
-	Message  bool `toml:"message" yaml:"message"`
+	Read     *bool `toml:"read" yaml:"read"`
+	Insert   *bool `toml:"insert" yaml:"insert"`
+	Update   *bool `toml:"update" yaml:"update"`
+	Delete   *bool `toml:"delete" yaml:"delete"`
+	Truncate *bool `toml:"truncate" yaml:"truncate"`
+	Message  *bool `toml:"message" yaml:"message"`
 }
 
 type AwsKinesisConfig struct {

spi/config/configuration_test.go

@@ -394,3 +394,117 @@ func Test_Config_Tags_Match_Between_Yaml_Toml(
 
 	recursiveCheck(configValue, "")
 }
+
+func Test_Config_Events_Values(
+	t *testing.T,
+) {
+
+	c := &Config{}
+	c.TimescaleDB.Events.Read = lo.ToPtr(false)
+	c.TimescaleDB.Events.Insert = lo.ToPtr(false)
+	c.TimescaleDB.Events.Update = lo.ToPtr(false)
+	c.TimescaleDB.Events.Delete = lo.ToPtr(false)
+	c.TimescaleDB.Events.Truncate = lo.ToPtr(false)
+	c.TimescaleDB.Events.Message = lo.ToPtr(false)
+	c.TimescaleDB.Events.Compression = lo.ToPtr(false)
+	c.TimescaleDB.Events.Decompression = lo.ToPtr(false)
+
+	c.PostgreSQL.Events.Read = lo.ToPtr(false)
+	c.PostgreSQL.Events.Insert = lo.ToPtr(false)
+	c.PostgreSQL.Events.Update = lo.ToPtr(false)
+	c.PostgreSQL.Events.Delete = lo.ToPtr(false)
+	c.PostgreSQL.Events.Truncate = lo.ToPtr(false)
+	c.PostgreSQL.Events.Message = lo.ToPtr(false)
+
+	assert.False(t, GetOrDefault(
+		c, PropertyHypertableEventsRead, true,
+	))
+	assert.False(t, GetOrDefault(
+		c, PropertyHypertableEventsInsert, true,
+	))
+	assert.False(t, GetOrDefault(
+		c, PropertyHypertableEventsUpdate, true,
+	))
+	assert.False(t, GetOrDefault(
+		c, PropertyHypertableEventsDelete, true,
+	))
+	assert.False(t, GetOrDefault(
+		c, PropertyHypertableEventsTruncate, true,
+	))
+	assert.False(t, GetOrDefault(
+		c, PropertyHypertableEventsCompression, true,
+	))
+	assert.False(t, GetOrDefault(
+		c, PropertyHypertableEventsDecompression, true,
+	))
+
+	assert.False(t, GetOrDefault(
+		c, PropertyPostgresqlEventsRead, true,
+	))
+	assert.False(t, GetOrDefault(
+		c, PropertyPostgresqlEventsInsert, true,
+	))
+	assert.False(t, GetOrDefault(
+		c, PropertyPostgresqlEventsUpdate, true,
+	))
+	assert.False(t, GetOrDefault(
+		c, PropertyPostgresqlEventsDelete, true,
+	))
+	assert.False(t, GetOrDefault(
+		c, PropertyPostgresqlEventsTruncate, true,
+	))
+
+	c.TimescaleDB.Events.Read = lo.ToPtr(true)
+	c.TimescaleDB.Events.Insert = lo.ToPtr(true)
+	c.TimescaleDB.Events.Update = lo.ToPtr(true)
+	c.TimescaleDB.Events.Delete = lo.ToPtr(true)
+	c.TimescaleDB.Events.Truncate = lo.ToPtr(true)
+	c.TimescaleDB.Events.Message = lo.ToPtr(true)
+	c.TimescaleDB.Events.Compression = lo.ToPtr(true)
+	c.TimescaleDB.Events.Decompression = lo.ToPtr(true)
+
+	c.PostgreSQL.Events.Read = lo.ToPtr(true)
+	c.PostgreSQL.Events.Insert = lo.ToPtr(true)
+	c.PostgreSQL.Events.Update = lo.ToPtr(true)
+	c.PostgreSQL.Events.Delete = lo.ToPtr(true)
+	c.PostgreSQL.Events.Truncate = lo.ToPtr(true)
+	c.PostgreSQL.Events.Message = lo.ToPtr(true)
+
+	assert.True(t, GetOrDefault(
+		c, PropertyHypertableEventsRead, false,
+	))
+	assert.True(t, GetOrDefault(
+		c, PropertyHypertableEventsInsert, false,
+	))
+	assert.True(t, GetOrDefault(
+		c, PropertyHypertableEventsUpdate, false,
+	))
+	assert.True(t, GetOrDefault(
+		c, PropertyHypertableEventsDelete, false,
+	))
+	assert.True(t, GetOrDefault(
+		c, PropertyHypertableEventsTruncate, false,
+	))
+	assert.True(t, GetOrDefault(
+		c, PropertyHypertableEventsCompression, false,
+	))
+	assert.True(t, GetOrDefault(
+		c, PropertyHypertableEventsDecompression, false,
+	))
+
+	assert.True(t, GetOrDefault(
+		c, PropertyPostgresqlEventsRead, false,
+	))
+	assert.True(t, GetOrDefault(
+		c, PropertyPostgresqlEventsInsert, false,
+	))
+	assert.True(t, GetOrDefault(
+		c, PropertyPostgresqlEventsUpdate, false,
+	))
+	assert.True(t, GetOrDefault(
+		c, PropertyPostgresqlEventsDelete, false,
+	))
+	assert.True(t, GetOrDefault(
+		c, PropertyPostgresqlEventsTruncate, false,
+	))
+}

tests/streamer_test.go

@@ -619,7 +619,7 @@ func (its *IntegrationTestSuite) TestCompressionEvents() {
 
 			ctx.AddSystemConfigConfigurator(testSink.SystemConfigConfigurator)
 			ctx.AddSystemConfigConfigurator(func(config *sysconfig.SystemConfig) {
-				config.TimescaleDB.Events.Compression = true
+				config.TimescaleDB.Events.Compression = lo.ToPtr(true)
 			})
 			return nil
 		}),
@@ -740,8 +740,8 @@ func (its *IntegrationTestSuite) TestCompressionPartialInsertEvents() {
 
 			ctx.AddSystemConfigConfigurator(testSink.SystemConfigConfigurator)
 			ctx.AddSystemConfigConfigurator(func(config *sysconfig.SystemConfig) {
-				config.TimescaleDB.Events.Compression = true
-				config.TimescaleDB.Events.Decompression = true
+				config.TimescaleDB.Events.Compression = lo.ToPtr(true)
+				config.TimescaleDB.Events.Decompression = lo.ToPtr(true)
 			})
 			return nil
 		}),
@@ -862,8 +862,8 @@ func (its *IntegrationTestSuite) TestDecompressionEvents() {
 
 			ctx.AddSystemConfigConfigurator(testSink.SystemConfigConfigurator)
 			ctx.AddSystemConfigConfigurator(func(config *sysconfig.SystemConfig) {
-				config.TimescaleDB.Events.Compression = true
-				config.TimescaleDB.Events.Decompression = true
+				config.TimescaleDB.Events.Compression = lo.ToPtr(true)
+				config.TimescaleDB.Events.Decompression = lo.ToPtr(true)
 			})
 			return nil
 		}),
@@ -984,8 +984,8 @@ func (its *IntegrationTestSuite) TestCompression_Decompression_SingleTransaction
 
 			ctx.AddSystemConfigConfigurator(testSink.SystemConfigConfigurator)
 			ctx.AddSystemConfigConfigurator(func(config *sysconfig.SystemConfig) {
-				config.TimescaleDB.Events.Compression = true
-				config.TimescaleDB.Events.Decompression = true
+				config.TimescaleDB.Events.Compression = lo.ToPtr(true)
+				config.TimescaleDB.Events.Decompression = lo.ToPtr(true)
 			})
 			return nil
 		}),