forked from airbytehq/airbyte
-
Notifications
You must be signed in to change notification settings - Fork 1
/
AirbyteMessageMigrator.java
78 lines (65 loc) · 3.45 KB
/
AirbyteMessageMigrator.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
/*
* Copyright (c) 2023 Airbyte, Inc., all rights reserved.
*/
package io.airbyte.commons.protocol;
import com.google.common.annotations.VisibleForTesting;
import io.airbyte.commons.protocol.migrations.AirbyteMessageMigration;
import io.airbyte.commons.protocol.migrations.MigrationContainer;
import io.airbyte.commons.version.Version;
import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
import java.util.List;
import java.util.Optional;
import java.util.Set;
/**
* AirbyteProtocol Message Migrator
*
* This class is intended to apply the transformations required to go from one version of the
* AirbyteProtocol to another.
*/
public class AirbyteMessageMigrator {
private final MigrationContainer<AirbyteMessageMigration<?, ?>> migrationContainer;
public AirbyteMessageMigrator(final List<AirbyteMessageMigration<?, ?>> migrations) {
migrationContainer = new MigrationContainer<>(migrations);
}
public void initialize() {
migrationContainer.initialize();
}
/**
* Downgrade a message from the most recent version to the target version by chaining all the
* required migrations
*/
public <PreviousVersion, CurrentVersion> PreviousVersion downgrade(final CurrentVersion message,
final Version target,
final Optional<ConfiguredAirbyteCatalog> configuredAirbyteCatalog) {
return migrationContainer.downgrade(message, target, (migration, msg) -> applyDowngrade(migration, msg, configuredAirbyteCatalog));
}
/**
* Upgrade a message from the source version to the most recent version by chaining all the required
* migrations
*/
public <PreviousVersion, CurrentVersion> CurrentVersion upgrade(final PreviousVersion message,
final Version source,
final Optional<ConfiguredAirbyteCatalog> configuredAirbyteCatalog) {
return migrationContainer.upgrade(message, source, (migration, msg) -> applyUpgrade(migration, msg, configuredAirbyteCatalog));
}
public Version getMostRecentVersion() {
return migrationContainer.getMostRecentVersion();
}
// Helper function to work around type casting
private static <PreviousVersion, CurrentVersion> PreviousVersion applyDowngrade(final AirbyteMessageMigration<PreviousVersion, CurrentVersion> migration,
final Object message,
final Optional<ConfiguredAirbyteCatalog> configuredAirbyteCatalog) {
return migration.downgrade((CurrentVersion) message, configuredAirbyteCatalog);
}
// Helper function to work around type casting
private static <PreviousVersion, CurrentVersion> CurrentVersion applyUpgrade(final AirbyteMessageMigration<PreviousVersion, CurrentVersion> migration,
final Object message,
final Optional<ConfiguredAirbyteCatalog> configuredAirbyteCatalog) {
return migration.upgrade((PreviousVersion) message, configuredAirbyteCatalog);
}
// Used for inspection of the injection
@VisibleForTesting
Set<String> getMigrationKeys() {
return migrationContainer.getMigrationKeys();
}
}