Documentation
¶
Index ¶
- Variables
- func EnsureTableExists(db *sql.DB, tableName string, row map[string]interface{}) error
- func EnsureTableExistsWorker(db *sql.DB, tasks chan map[string]interface{}, errorsChan chan error, ...)
- func ReadCSV(fileName string) ([]byte, error)
- func TransformDataToBSON(data interface{}) ([]bson.M, error)
- func ValidateJSONData(data string) (interface{}, error)
- func ValidateYAMLData(data []byte) (interface{}, error)
- func WriteCSV(fileName string, data []byte) error
- type CSVDestination
- type CSVSource
- type DynamoDBDestination
- type DynamoDBSource
- type FTPDestination
- type FTPSource
- type FirebaseDestination
- type FirebaseSource
- type JSONDestination
- type JSONSource
- type KafkaDestination
- type KafkaSource
- type MockDynamoDB
- type MongoDBDestination
- type MongoDBSource
- type PostgreSQLDestination
- type PostgreSQLSource
- type RabbitMQDestination
- type RabbitMQSource
- type SFTPDestination
- type SFTPSource
- type WebSocketDestination
- type WebSocketSource
- type YAMLDestination
- type YAMLSource
Constants ¶
This section is empty.
Variables ¶
var ( ErrFTPConnectionFailed = errors.New("failed to connect to FTP server") ErrFTPLoginFailed = errors.New("failed to login to FTP server") ErrFTPFileNotFound = errors.New("file not found on FTP server") ErrFTPFileUploadFailed = errors.New("failed to upload file to FTP server") )
Predefined FTP errors
Functions ¶
func EnsureTableExists ¶
EnsureTableExists enqueues table creation tasks and processes them concurrently.
func EnsureTableExistsWorker ¶
func EnsureTableExistsWorker(db *sql.DB, tasks chan map[string]interface{}, errorsChan chan error, done chan bool)
EnsureTableExistsWorker processes table creation tasks.
func TransformDataToBSON ¶
func ValidateJSONData ¶
ValidateJSONData validates, sanitizes, and unmarshals JSON data
func ValidateYAMLData ¶
ValidateYAMLData unmarshals and validates the YAML data.
Types ¶
type CSVDestination ¶
type CSVDestination struct {
CSVDestinationFileName string `json:"csv_destination_file_name"`
}
CSVDestination struct represents the configuration for publishing messages to CSV.
func (CSVDestination) SendData ¶
func (r CSVDestination) SendData(data interface{}, req interfaces.Request) error
SendData writes data to a CSV file concurrently.
type CSVSource ¶
type CSVSource struct {
CSVSourceFileName string `json:"csv_source_file_name"`
}
CSVSource struct represents the configuration for consuming messages from CSV.
type DynamoDBDestination ¶
type DynamoDBDestination struct {
TableName string `json:"table_name"`
Region string `json:"region"`
}
DynamoDBDestination represents the configuration for writing data to DynamoDB.
func (DynamoDBDestination) SendData ¶
func (d DynamoDBDestination) SendData(data interface{}, req interfaces.Request) error
SendData writes data to the target DynamoDB table in the specified region.
type DynamoDBSource ¶
DynamoDBSource represents the configuration for reading data from DynamoDB.
func (DynamoDBSource) FetchData ¶
func (d DynamoDBSource) FetchData(req interfaces.Request) (interface{}, error)
FetchData retrieves data from the source DynamoDB table in the specified region.
type FTPDestination ¶
type FTPDestination struct {
URL string `json:"url"`
User string `json:"user"`
Password string `json:"password"`
FTPFILEPATH string `json:"file_path"`
}
FTPDestination implements the DataDestination interface
func (FTPDestination) SendData ¶
func (f FTPDestination) SendData(data interface{}, req interfaces.Request) error
SendData sends data to an FTP server
type FTPSource ¶
type FTPSource struct {
URL string `json:"url"`
User string `json:"user"`
Password string `json:"password"`
FTPFILEPATH string `json:"file_path"`
}
FTPSource implements the DataSource interface
type FirebaseDestination ¶
type FirebaseDestination struct {
CredentialFileAddr string `json:"firebase_credential_file"`
Collection string `json:"firebase_collection"`
Document string `json:"firebase_document"`
}
func (FirebaseDestination) SendData ¶
func (f FirebaseDestination) SendData(data interface{}, req interfaces.Request) error
type FirebaseSource ¶
type FirebaseSource struct {
CredentialFileAddr string `json:"firebase_credential_file"`
Collection string `json:"firebase_collection"`
Document string `json:"firebase_document"`
}
func (FirebaseSource) FetchData ¶
func (f FirebaseSource) FetchData(req interfaces.Request) (interface{}, error)
type JSONDestination ¶
type JSONDestination struct {
Filename string `json:"json_output_filename"`
}
func (JSONDestination) SendData ¶
func (j JSONDestination) SendData(data interface{}, req interfaces.Request) error
SendData writes JSON data to a destination file
type JSONSource ¶
type JSONSource struct {
Data string `json:"json_source_data"`
}
func (JSONSource) FetchData ¶
func (j JSONSource) FetchData(req interfaces.Request) (interface{}, error)
FetchData retrieves and processes JSON source data
type KafkaDestination ¶
type KafkaDestination struct {
URL string `json:"producer_url"`
Topic string `json:"producer_topic"`
}
KafkaDestination struct represents the configuration for publishing messages to Kafka.
func (KafkaDestination) SendData ¶
func (k KafkaDestination) SendData(data interface{}, req interfaces.Request) error
SendData connects to Kafka and publishes data to the specified topic concurrently.
type KafkaSource ¶
KafkaSource struct represents the configuration for consuming messages from Kafka.
func (KafkaSource) FetchData ¶
func (k KafkaSource) FetchData(req interfaces.Request) (interface{}, error)
FetchData connects to Kafka, retrieves data, and processes it concurrently.
type MockDynamoDB ¶
type MockDynamoDB struct {
dynamodbiface.DynamoDBAPI
}
MockDynamoDB is a mock struct for simulating DynamoDB operations.
func (*MockDynamoDB) PutItem ¶
func (m *MockDynamoDB) PutItem(input *dynamodb.PutItemInput) (*dynamodb.PutItemOutput, error)
func (*MockDynamoDB) Scan ¶
func (m *MockDynamoDB) Scan(input *dynamodb.ScanInput) (*dynamodb.ScanOutput, error)
type MongoDBDestination ¶
type MongoDBDestination struct {
ConnString string `json:"target_mongodb_conn_string"`
Database string `json:"target_mongodb_database"`
Collection string `json:"target_mongodb_collection"`
}
MongoDBDestination struct represents the configuration for publishing messages to MongoDB.
func (MongoDBDestination) SendData ¶
func (m MongoDBDestination) SendData(data interface{}, req interfaces.Request) error
SendData connects to MongoDB and publishes data to the specified collection.
type MongoDBSource ¶
type MongoDBSource struct {
ConnString string `json:"source_mongodb_conn_string"`
Database string `json:"source_mongodb_database"`
Collection string `json:"source_mongodb_collection"`
}
MongoDBSource struct represents the configuration for consuming messages from MongoDB.
func (MongoDBSource) FetchData ¶
func (m MongoDBSource) FetchData(req interfaces.Request) (interface{}, error)
FetchData connects to MongoDB, retrieves data, and returns it.
type PostgreSQLDestination ¶
type PostgreSQLDestination struct {
ConnString string `json:"postgresql_target_conn_string"`
}
PostgreSQLDestination struct represents the configuration for publishing messages to PostgreSQL.
func (PostgreSQLDestination) SendData ¶
func (p PostgreSQLDestination) SendData(data interface{}, req interfaces.Request) error
SendData connects to PostgreSQL and publishes data to the specified table.
type PostgreSQLSource ¶
type PostgreSQLSource struct {
ConnString string `json:"postgresql_source_conn_string"`
}
PostgreSQLSource struct represents the configuration for consuming messages from PostgreSQL.
func (PostgreSQLSource) FetchData ¶
func (p PostgreSQLSource) FetchData(req interfaces.Request) (interface{}, error)
FetchData connects to PostgreSQL, retrieves data, and returns it.
type RabbitMQDestination ¶
type RabbitMQDestination struct {
URL string `json:"rabbitmq_output_url"`
QueueName string `json:"rabbitmq_output_queue_name"`
}
RabbitMQDestination struct represents the configuration for publishing messages to RabbitMQ.
func (RabbitMQDestination) SendData ¶
func (r RabbitMQDestination) SendData(data interface{}, req interfaces.Request) error
SendData connects to RabbitMQ and publishes data to the specified queue.
type RabbitMQSource ¶
type RabbitMQSource struct {
URL string `json:"rabbitmq_input_url"`
QueueName string `json:"rabbitmq_input_queue_name"`
}
RabbitMQSource struct represents the configuration for consuming messages from RabbitMQ.
func (RabbitMQSource) FetchData ¶
func (r RabbitMQSource) FetchData(req interfaces.Request) (interface{}, error)
FetchData connects to RabbitMQ, retrieves data, and processes it concurrently.
type SFTPDestination ¶
type SFTPDestination struct {
URL string `json:"url"`
User string `json:"user"`
Password string `json:"password"`
SFTPFILEPATH string `json:"file_path"`
}
SFTPDestination implements the DataDestination interface
func (SFTPDestination) SendData ¶
func (s SFTPDestination) SendData(data interface{}, req interfaces.Request) error
SendData sends data to an SFTP server concurrently
type SFTPSource ¶
type SFTPSource struct {
URL string `json:"url"`
User string `json:"user"`
Password string `json:"password"`
SFTPFILEPATH string `json:"file_path"`
}
SFTPSource implements the DataSource interface
func (SFTPSource) FetchData ¶
func (s SFTPSource) FetchData(req interfaces.Request) (interface{}, error)
FetchData fetches data from an SFTP server concurrently
type WebSocketDestination ¶
type WebSocketDestination struct {
URL string `json:"websocket_dest_url"`
}
WebSocketDestination struct represents the configuration for publishing messages to WebSocket.
func (WebSocketDestination) SendData ¶
func (ws WebSocketDestination) SendData(data interface{}, req interfaces.Request) error
SendData connects to WebSocket and publishes data to the specified WebSocket server.
type WebSocketSource ¶
type WebSocketSource struct {
URL string `json:"websocket_source_url"`
}
WebSocketSource struct represents the configuration for consuming messages from WebSocket.
func (WebSocketSource) FetchData ¶
func (ws WebSocketSource) FetchData(req interfaces.Request) (interface{}, error)
FetchData connects to WebSocket, retrieves data, and passes it through validation and transformation pipelines.
type YAMLDestination ¶
type YAMLDestination struct {
FilePath string `json:"yaml_output_file_path"`
}
YAMLDestination struct represents the configuration for writing data to a YAML file.
func (YAMLDestination) SendData ¶
func (y YAMLDestination) SendData(data interface{}, req interfaces.Request) error
SendData writes the provided data to a YAML destination file.
type YAMLSource ¶
type YAMLSource struct {
FilePath string `json:"yaml_source_file_path"`
}
YAMLSource struct represents the configuration for reading data from a YAML file.
func (YAMLSource) FetchData ¶
func (y YAMLSource) FetchData(req interfaces.Request) (interface{}, error)
FetchData reads and processes data from a YAML source file.
Source Files