Référence des Composants

Vue d’ensemble

Référence complète de tous les composants GoCamel disponibles. Les composants fournissent la connectivité à divers systèmes et services.

Composants Principaux

Direct

Routage synchrone en mémoire entre routes dans le même contexte.

// Consumer : reçoit depuis un endpoint direct
// Producer : envoie vers un endpoint direct
builder.From("direct:start").To("direct:process")

Les endpoints sont identifiés par leur nom uniquement — les paramètres de requête sont ignorés pour l’identité, donc direct:start et direct:start?x=1 correspondent au même endpoint. Un seul consumer par endpoint est autorisé.

Timer

Déclenchement périodique simple.

builder.From("timer:tick?period=5s")

Composants de Transfert de Fichiers

File

Opérations sur le système de fichiers local.

// Consumer (lecture)
builder.From("file://input?delete=true")

// Producer (écriture)
builder.To("file://output")

En-têtes

Définis par le Consumer

• CamelFileName : Nom relatif du fichier
• CamelFilePath : Chemin absolu ou complet du fichier
• CamelFileLength : Taille du fichier en octets
• CamelFileLastModified : Date et heure de dernière modification (time.Time)

Consommés par le Producer

• CamelFileName : Surcharge le nom du fichier lorsque l’URI cible pointe vers un répertoire. Doit être un chemin relatif : nom vide, chemin absolu, volume Windows et composants .. qui sortent du répertoire de l’endpoint sont rejetés.

Le consumer ignore les liens symboliques et refuse les fichiers de plus de 32 Mio (DefaultMaxBodySize).

Verrou de lecture (Read Lock)

GoCamel fournit plusieurs stratégies de verrou de lecture pour s’assurer que les fichiers sont complètement écrits et exclusifs avant traitement :

• readLock=changed : Le consumer échantillonne la taille et la date de modification sur readLockCheckInterval et lit dès que deux échantillons consécutifs coïncident dans la limite de readLockTimeout.
• readLock=rename : Renomme temporairement le fichier sous un nom exclusif (.camelExclusiveReadLock) pendant le traitement.
• readLock=markerFile : Crée un fichier marqueur .camelLock pendant le traitement et le supprime une fois terminé.
• readLock=none : Désactive toute vérification de verrou.

FTP / FTPS

Transfert de fichiers via le protocole FTP.

// Consumer
builder.From("ftp://host:21/incoming?username=admin&readLock=changed")

// Producer
builder.To("ftp://host:21/outgoing?binary=true")

Variables d’environnement :

• FTP_USERNAME - Nom d’utilisateur
• FTP_PASSWORD - Mot de passe

Avec fileExist=Append, le producer télécharge le fichier distant existant puis renvoie l’envoi de la concaténation. Un téléchargement partiel fait échouer l’envoi plutôt que de renvoyer silencieusement un contenu tronqué.

Lorsque le chemin de l’URI est un répertoire, CamelFileName est joint avec JoinRemotePath : noms absolus et traversée de répertoires sont rejetés.

SFTP

Transfert de fichiers sécurisé via SSH.

builder.From("sftp://host:22/data?username=scott&readLock=changed&readLockTimeout=15s")

Méthodes d’authentification :

• Par mot de passe : via le paramètre password ou la variable d’environnement SFTP_PASSWORD
• Par clé : via le paramètre privateKeyFile ou SFTP_PRIVATE_KEY_FILE

Comme pour FTP, fileExist=Append concatène au fichier distant existant et fait échouer l’envoi si sa lecture échoue en cours de route. CamelFileName est confiné comme pour FTP (JoinRemotePath).

SMB

Accès aux partages Windows/Samba.

builder.From("smb://server/share/folder?username=user&readLock=changed")

La sémantique de fileExist=Append est identique à celle de FTP/SFTP. CamelFileName est confiné avec JoinRemotePath (noms absolus et .. rejetés ; le producer n’utilise pas filepath.Join, qui jetterait le chemin du partage si le header était absolu).

Composants Réseau

HTTP

Support serveur et client HTTP.

// Consumer (serveur HTTP) — écoute en clair uniquement
builder.From("http://localhost:8080/api")

// Producer (client HTTP)
builder.To("http://example.com/webhook")

Le scheme https est producer uniquement. From("https://...") échoue à la création du consumer : le listener est du HTTP en clair. Terminez le TLS devant un consumer http:// (reverse proxy, HTTPComponent.SetMiddleware, etc.).

Headers posés par le consumer

Le producer envoie en POST sauf si CamelHttpMethod est posé sur l’exchange (un pont From(http).To(http) transmet donc la méthode entrante).

Les headers de la requête entrante sont aussi copiés sur l’exchange (voir headers de réponse ci-dessous).

Authentification, TLS, limitation de débit (middleware consumer)

Le consumer HTTP est livré sans authentification. Ne l’exposez à un réseau non fiable que derrière un middleware qui applique le modèle de sécurité dont vous avez besoin (vérification de token, mTLS, OAuth, allowlist d’IP, rate limit, …). Installez le middleware sur le HTTPComponent avant le démarrage de la route :

http := gocamel.NewHTTPComponent()
http.SetMiddleware(func(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        if r.Header.Get("Authorization") != "Bearer "+os.Getenv("API_TOKEN") {
            http.Error(w, "non autorisé", http.StatusUnauthorized)
            return
        }
        next.ServeHTTP(w, r)
    })
})
ctx.AddComponent("http", http)

Le middleware enveloppe chaque handler installé par un consumer http:// dans ce contexte.

Sanitisation des headers sortants

Le producer HTTP (To) et le chemin de réponse du consumer rejettent les noms et valeurs de header contenant \r ou \n (CRLF) pour empêcher le HTTP response splitting (CWE-113). Un exchange qui tente de définir un tel header reçoit une erreur de la part de Send().

Headers de réponse

Le corps de la réponse provient de Out (ou de In si aucun Out n’a été défini, selon la sémantique InOut de Camel). Les corps []byte et string sont écrits tels quels ; tout autre corps non nil est écrit en utilisant sa représentation %v.

Comme le consumer copie tous les headers de la requête entrante sur l’exchange, une route qui ne définit pas Out répondrait avec les headers de la requête elle-même. Deux filtres l’en empêchent :

• Les headers entrants inchangés ne sont pas renvoyés. Un header qui revient identique à celui reçu est écarté : Cookie et Authorization ne sont donc jamais réfléchis vers le client. Un header posé ou modifié par la route est toujours émis.
• Les headers hop-by-hop et de cadrage ne sont jamais émis (Connection, Keep-Alive, Transfer-Encoding, Content-Length, Host, …). Renvoyer le Content-Length de la requête corrompait le cadrage de la réponse.

Le même filtre s’applique au producer : un pont From(http://...) → To(http://...) ne transmet donc pas le Content-Length ni le Host entrants au service amont.

Le producer transmet par défaut tous les autres headers entrants, y compris Cookie et Authorization — le comportement normal d’un client HTTP générique qui appelle un service amont authentifié. Dans un scénario de pont/proxy où cela fuirait les identifiants d’un client vers un amont tiers, supprimez-les explicitement :

http := gocamel.NewHTTPComponent()
http.SetSkipAuthorizationHeaders(true)

Une fois activée, Authorization et Cookie sont retirés de toute requête sortante produite par ce composant.

Statut de la réponse

Une route fixe le code de statut via le header CamelHttpResponseCode sur le message de réponse (l’ancien header Status-Code reste accepté). Les valeurs hors de 100–599 sont ignorées :

builder.From("http://localhost:8080/api").
    ProcessFunc(func(e *gocamel.Exchange) error {
        e.GetOut().SetHeader(gocamel.CamelHttpResponseCode, 201)
        e.GetOut().SetHeader("Content-Type", "application/json")
        e.GetOut().SetBody(`{"created":true}`)
        return nil
    })

Gestion des erreurs du producer

Une réponse amont de statut ≥ 300 est une erreur (comportement par défaut d’Apache Camel). L’erreur enveloppe gocamel.ErrHTTPStatus, et le corps ainsi que les headers de la réponse restent publiés sur Out pour inspection :

if errors.Is(err, gocamel.ErrHTTPStatus) {
    code, _ := exchange.GetOut().GetHeader(gocamel.CamelHttpResponseCode)
    // code est un int ; le corps contient la charge utile d'erreur amont
}

Pour considérer toute réponse comme un succès et traiter le statut soi-même :

http := gocamel.NewHTTPComponent()
http.SetThrowExceptionOnFailure(false)

Le producer renseigne également CamelHttpResponseCode (int) et CamelHttpResponseText (ex. "500 Internal Server Error") sur Out.

Durcissement

• Le serveur embarqué définit un ReadHeaderTimeout de 10 secondes (attaques d’en-têtes lents), un ReadTimeout de 60 secondes (attaques de corps lents) et un IdleTimeout de 60 secondes (connexions keep-alive inactives).

• Les corps de requête et de réponse sont lus au travers d’une limite de taille — DefaultMaxBodySize (32 Mio) — afin qu’une charge utile surdimensionnée ne puisse pas épuiser la mémoire du processus (CWE-400). Une requête trop volumineuse reçoit un 413 Request Entity Too Large. Ajustez ou désactivez la limite par composant :

  http := gocamel.NewHTTPComponent()
  http.SetMaxBodySize(256 << 20) // 256 Mio ; <= 0 désactive la limite

• Les erreurs de route ne sont pas renvoyées au client : le handler répond par un 500 internal server error générique et journalise le détail, qui peut contenir du texte SQL, des chemins de fichiers ou des messages de driver.

Arrêt

La goroutine du consumer est suivie par un WaitGroup interne et Stop() attend que http.Server.Shutdown se termine (borné par un délai interne de 30 secondes) avant de retourner. Stop() est idempotent.

Net (TCP/UDP)

Sockets TCP et UDP bruts, construits sur la bibliothèque standard — l’équivalent Go des composants netty/mina d’Apache Camel. Le consumer écoute et alimente la route avec les messages entrants ; le producer se connecte et envoie le corps du message.

// Consumer (serveur) : requête-réponse sur TCP
builder.From("net:tcp://localhost:9090")

// Consumer (serveur) : UDP sans accusé de réception
builder.From("net:udp://localhost:9090?sync=false")

// Producer (client)
builder.To("net:tcp://localhost:9090")

Formes d’URI — net:tcp://hôte:port et net:tcp:hôte:port sont acceptées ; idem pour udp. Un hôte omis prend la valeur localhost. Le port 0 demande un port éphémère à l’OS : lisez l’adresse réellement liée via NetConsumer.Address().

Cadrage. TCP est un flux d’octets : le composant a donc besoin d’une limite de message. Avec textline=true (défaut), chaque ligne terminée par un saut de ligne est un message (le \r\n ou \n final est retiré). Avec textline=false, chaque lecture est un message. Les datagrammes UDP n’ont pas besoin de cadrage : chaque datagramme est un message.

// Serveur écho requête-réponse
builder.From("net:tcp://localhost:9090").
    ProcessFunc(func(e *gocamel.Exchange) error {
        body, _ := e.GetIn().GetBody().([]byte)
        e.GetOut().SetBody([]byte("echo: " + string(body)))
        return nil
    })

Headers. Le consumer définit CamelNetRemoteAddress et CamelNetLocalAddress (hôte:port) sur le message In.

Durcissement

• Les lectures de messages sont bornées par bufferSize. Un message textline plus long que la limite est une erreur de protocole : la connexion est fermée plutôt que de mettre en tampon des données illimitées (CWE-400). Les datagrammes UDP plus grands que la limite sont tronqués.
• La lecture de la réponse du producer en mode sync est bornée par bufferSize et par le délai timeout, afin qu’un pair muet ne puisse pas bloquer la route indéfiniment.
• Le consumer fixe un readTimeout (défaut 60 s) sur chaque connexion afin qu’un client de type slowloris ne puisse pas monopoliser une goroutine indéfiniment ; mettez-le à 0 pour le désactiver.
• Les paniques dans les processeurs de route sont contenues (ProcessSafely) : une route défectueuse échoue son exchange, jamais le consumer.

Composants de Messagerie

Telegram

Intégration de l’API Telegram Bot pour la réception et l’envoi de messages.

// Requis : définir la variable d'environnement TELEGRAM_AUTHORIZATIONTOKEN
ctx.AddComponent("telegram", gocamel.NewTelegramComponent())

// Consumer (mode webhook/polling)
builder.From("telegram:bots").Log("${body}")

// Producer (envoi de messages)
builder.To("telegram:bots")

Variables d’environnement :

• TELEGRAM_AUTHORIZATIONTOKEN - Token de l’API Bot

NATS

Messagerie asynchrone orientée événements avec NATS. Publisher et subscriber haute performance avec le client Go NATS.

ctx.AddComponent("nats", gocamel.NewNATSComponent())

// Consumer : subscription asynchrone réactive
builder.From("nats://orders?servers=nats://localhost:4222").
    ProcessFunc(func(e *gocamel.Exchange) error {
        // La charge utile NATS est reçue comme un slice d'octets
        payload := e.GetIn().GetBody().([]byte)
        fmt.Printf("Commande reçue : %s\n", string(payload))
        return nil
    })

// Producer : publier vers un sujet
builder.To("nats://orders?servers=nats://localhost:4222")

// Surcharger dynamiquement le sujet via un header
exchange.In.SetHeader(gocamel.CamelNATSSubject, "urgent-orders")

Headers d’Exchange :

• CamelNATSSubject - Sujet du message reçu ou sujet cible pour la publication.

Kafka

Diffusion d’événements distribuée à haut débit avec Apache Kafka via segmentio/kafka-go (pur Go).

import "gitlab.com/tranchida/gocamel/components/kafka"

ctx.AddComponent("kafka", kafka.NewKafkaComponent())

// Consommateur avec consumer group
builder.From("kafka:orders?brokers=localhost:9092&groupId=billing-service").
    Log("Message reçu : ${body}")

// Producteur avec partition key dynamique (nécessite allowHeaderOverride=true)
builder.From("direct:publish").
    SetHeader("CamelKafkaPartitionKey", "key-123").
    To("kafka:orders?brokers=localhost:9092&allowHeaderOverride=true")

Headers d’Exchange :

• CamelKafkaTopic - Surcharge (avec allowHeaderOverride=true) ou lecture du topic cible.
• CamelKafkaPartitionKey / CamelKafkaKey - Clé de partitionnement Kafka (honorée avec allowHeaderOverride=true).
• CamelKafkaPartition - Index de la partition.
• CamelKafkaOffset - Offset du message.
• CamelKafkaTimestamp - Horodatage du message.

⚠️ Sécurité : le topic et la clé configurés dans l’URI constituent une configuration de confiance. Les surcharges par headers sont ignorées sauf si l’endpoint active explicitement allowHeaderOverride=true. Ne l’activez que si les headers ne peuvent pas être influencés par une entrée non fiable.

Les headers sortants de l’exchange sont filtrés avant d’atteindre le broker : Authorization, Cookie, Proxy-Authorization, Set-Cookie et WWW-Authenticate (insensible à la casse) ne sont jamais propagés, afin qu’un producteur en aval du consommateur HTTP ne fuite pas les identifiants du client. Définissez forwardHeaders pour une liste explicite si vous souhaitez un contrôle plus strict.

Quand un groupId est configuré, chaque message traité avec succès est committé auprès du groupe de consommateurs. Un échec de commit est journalisé (avec topic/partition/offset) plutôt qu’ignoré silencieusement : le message sera redélivré après un rééquilibrage ou un redémarrage, et la ligne de journal est le seul signal de ce traitement dupliqué.

gRPC

Intégration client et serveur RPC haute performance via google.golang.org/grpc.

import "gitlab.com/tranchida/gocamel/components/grpc"

ctx.AddComponent("grpc", grpc.NewGRPCComponent())

// Serveur Consommateur : route les appels RPC entrants dans GoCamel
builder.From("grpc://0.0.0.0:50051/helloworld.Greeter/SayHello").
    SetBody("Réponse depuis le serveur gRPC GoCamel")

// Client Producteur : invoque des méthodes RPC distantes
builder.From("direct:invoke").
    To("grpc://localhost:50051/helloworld.Greeter/SayHello?insecure=true")

Headers d’Exchange :

• CamelGrpcMethod - Méthode gRPC invoquée.
• CamelGrpcStatusCode - Code d’état gRPC (ex: 0 pour OK).
• CamelGrpcStatusDescription - Message descriptif du statut gRPC.

⚠️ Sécurité : les headers sortants de l’exchange sont filtrés avant d’être envoyés comme métadonnées RPC : Authorization, Cookie, Proxy-Authorization, Set-Cookie et WWW-Authenticate (insensible à la casse) ne sont jamais propagés, afin qu’un producteur en aval du consommateur HTTP ne fuite pas les identifiants du client. Définissez forwardHeaders pour une liste explicite si vous souhaitez un contrôle plus strict.

Redis

Envoi de commandes (SET, GET, DEL, PUBLISH) ou abonnement aux canaux Redis Pub/Sub.

ctx.AddComponent("redis", gocamel.NewRedisComponent())

// Consumer : s'abonner à un canal Pub/Sub
builder.From("redis://localhost:6379?subscribe=true&channel=mychannel").
    Log("Message Pub/Sub reçu : ${body}")

// Producer : exécuter des commandes
builder.To("redis://localhost:6379?command=SET&key=mykey")

Commandes supportées :

• SET : Enregistre le corps de l’exchange comme valeur de key.
• GET : Récupère la valeur de key et l’écrit dans Out.Body.
• DEL : Supprime key et écrit le nombre de clés affectées dans Out.Body.
• PUBLISH : Publie le corps de l’exchange vers le canal/clé spécifié.

Headers d’Exchange (surcharge des paramètres URI) :

• CamelRedisCommand - La commande Redis à exécuter (ex. "GET").
• CamelRedisKey - La clé Redis cible.
• CamelRedisChannel - Canal Pub/Sub cible pour la publication ou canal source pour l’abonnement.

Dépôt Idempotent Redis

Le Dépôt Idempotent Redis fournit une implémentation distribuée de l’interface IdempotentRepository, permettant à plusieurs instances GoCamel dans un cluster de se coordonner et d’éviter le traitement de messages dupliqués.

import (
    "time"
    "github.com/redis/go-redis/v9"
)

rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
// Préfixe de clé "idempotent:" avec TTL d'expiration de 24 heures
repo := gocamel.NewRedisIdempotentRepository(rdb, "idempotent:", 24*time.Hour)

builder.From("direct:start").
    IdempotentConsumer("${header.MessageId}", repo).
    To("direct:process")

Go Channel

Composant d’intégration intra-processus natif et ultra-rapide utilisant les canaux Go natifs. Il permet de découpler les segments de route de manière asynchrone au sein du même processus.

ctx.AddComponent("chan", gocamel.NewChanComponent())

// Consumer : traite les exchanges de manière asynchrone depuis le canal
builder.From("chan:orders?bufferSize=500").
    To("direct:process-order")

// Producer : écrit une copie de l'exchange dans le canal de manière asynchrone
builder.To("chan:orders")

Notes :

• Les endpoints sont mis en cache par nom de canal : chaque producer et consumer utilisant le même nom partage un endpoint, et un seul consumer par canal est autorisé (un second From("chan:orders") échoue au démarrage).
• Les options URI sont appliquées lorsque l’endpoint est créé pour la première fois ; les références ultérieures avec des options différentes émettent un avertissement et réutilisent l’endpoint existant.

Composants IA

OpenAI

Intégration de l’API OpenAI pour ChatGPT/GPT-4. Tout endpoint compatible OpenAI est pris en charge via l’option baseURL (OpenRouter, Groq, Together, Mistral, vLLM, LM Studio, couche de compatibilité OpenAI d’Ollama, …).

ctx.AddComponent("openai", gocamel.NewOpenAIComponent())

// Producer uniquement - envoyer une requête de chat completion
endpoint, _ := ctx.CreateEndpoint("openai:chat?model=gpt-4")
producer, _ := endpoint.CreateProducer()

exchange := gocamel.NewExchange(context.Background())
exchange.GetIn().SetBody("Bonjour, comment allez-vous ?")
producer.Send(exchange)

fmt.Println(exchange.GetOut().GetBody()) // Réponse de l'IA

Utiliser un fournisseur compatible OpenAI :

// Exemple OpenRouter : router vers openai/gpt-4o via la passerelle OpenRouter
endpoint, _ := ctx.CreateEndpoint(
    "openai:chat?model=openai/gpt-4o" +
        "&baseURL=https://openrouter.ai/api/v1" +
        "&apiKey=$OPENROUTER_API_KEY")

Variables d’environnement :

• OPENAI_AUTHORIZATIONTOKEN ou OPENAI_API_KEY - Clé API

Le timeout HTTP du SDK est de 60 secondes. Bornez davantage les appels longs avec exchange.Context.

Anthropic

Intégration native de l’API Messages d’Anthropic (Claude). Contrairement au composant OpenAI, c’est un connecteur dédié à la forme native de l’API Claude — paramètre system séparé, blocs de contenu, en-têtes de cache de prompt et raisons d’arrêt spécifiques.

ctx.AddComponent("anthropic", gocamel.NewAnthropicComponent())

// Producer uniquement - envoyer une requête Messages
endpoint, _ := ctx.CreateEndpoint(
    "anthropic:messages?model=claude-sonnet-4-5" +
        "&apiKey=$ANTHROPIC_API_KEY" +
        "&maxTokens=1024" +
        "&system=Tu es un assistant concis")
producer, _ := endpoint.CreateProducer()

exchange := gocamel.NewExchange(context.Background())
exchange.GetIn().SetBody("Explique EIP en une phrase.")
producer.Send(exchange)

fmt.Println(exchange.GetOut().GetBody()) // Réponse de Claude

Variables d’environnement :

• ANTHROPIC_API_KEY - Clé API (recommandé plutôt que l’intégrer dans l’URI)

Le timeout HTTP du SDK est de 60 secondes.

En-têtes de surcharge par message (message In) :

En-têtes Out (définis sur la réponse) :

Ollama

Intégration native avec un serveur Ollama local via son client Go officiel (github.com/ollama/ollama/api). Trois modes sont sélectionnés par le chemin de l’URI :

• ollama:chat (défaut) — API de chat multi-tours (/api/chat)
• ollama:generate — génération one-shot (/api/generate)
• ollama:embed — embeddings (/api/embed)

ctx.AddComponent("ollama", gocamel.NewOllamaComponent())

// Discuter avec un modèle local
endpoint, _ := ctx.CreateEndpoint(
    "ollama:chat?model=llama3.1&host=http://localhost:11434")
producer, _ := endpoint.CreateProducer()

exchange := gocamel.NewExchange(context.Background())
exchange.GetIn().SetBody("Résume EIP en une phrase.")
producer.Send(exchange)

fmt.Println(exchange.GetOut().GetBody()) // Réponse du modèle local

Générer des embeddings localement :

endpoint, _ := ctx.CreateEndpoint(
    "ollama:embed?model=nomic-embed-text&host=http://localhost:11434")
// exchange.GetOut().GetBody() retourne []float32

Variables d’environnement :

• OLLAMA_HOST - hôte par défaut lorsque l’option URI host est omise

En-têtes de surcharge par message (message In) :

En-têtes Out (définis sur la réponse) :

Le composant Ollama n’expose pas les opérations de cycle de vie des modèles (Pull/List/Delete). Utilisez la CLI ollama ou l’API HTTP directement pour la gestion des modèles ; le producer GoCamel est limité à l’inférence.

Composants d’Ordonnancement

Cron

Ordonnancement avancé avec expressions cron ou intervalles simples.

ctx.AddComponent("cron", gocamel.NewCronComponent())

// Déclencheur cron (6 champs, incluant les secondes)
builder.From("cron://group/job?cron=0+*+*+*+*+*")

// Déclencheur à intervalle simple
builder.From("cron://poller?trigger.repeatInterval=5000")

Format d’expression cron (6 champs) :

second minute hour day month dayOfWeek
0 * * * * *  # Chaque minute
0 */5 * * * * # Toutes les 5 minutes
cron=0 0 12 * * *  # Chaque jour à midi

Headers d’Exchange :

• fireTime - Heure d’exécution du déclencheur
• nextFireTime - Prochaine heure planifiée
• triggerName - Identifiant du déclencheur

À l’appel de Stop(), un job dont le triggerStartDelay n’est pas encore écoulé n’est jamais enregistré sur le scheduler partagé : la demande d’arrêt gagne la course avec l’enregistrement différé, aucun job orphelin ne peut donc rester à se déclencher indéfiniment.

Composants de Messagerie Électronique

SMTP/SMTPS (Envoi)

Envoi d’emails via SMTP.

builder.To("smtps://smtp.gmail.com:465?to=recipient@example.com&subject=Bonjour")

IMAP/IMAPS (Réception)

Réception d’emails via IMAP avec support IDLE.

builder.From("imaps://imap.gmail.com:993?folderName=INBOX&idle=true")

POP3/POP3S (Réception)

Réception d’emails via POP3.

builder.From("pop3s://pop.gmail.com:995?username=user&password=pass")

Le consumer POP3 accepte la même option maxMessageSize qu’IMAP (défaut 0 = illimité) pour borner la taille d’un message téléchargé.

Avec disconnect=false (défaut), la connexion POP3 est conservée entre les polls, comme pour le consumer IMAP. Définissez disconnect=true pour vous ré-authentifier à chaque poll (utile avec les serveurs qui coupent les connexions inactives).

Composants de Base de Données

SQL

Exécution de requêtes SQL via database/sql.

import (
    "database/sql"
    _ "modernc.org/sqlite"
)

db, _ := sql.Open("sqlite", "./app.db")

sqlComp := gocamel.NewSQLComponent()
sqlComp.RegisterDataSource("appdb", db)
ctx.AddComponent("sql", sqlComp)

// SELECT -> Out.Body = []map[string]any
builder.From("direct:list").
    To("sql://appdb?query=SELECT+id,name+FROM+users").
    Log("${body}")

// SELECT une seule ligne -> Out.Body = map[string]any
builder.From("direct:one").
    SetHeader(gocamel.SqlParameters, []any{42}).
    To("sql://appdb?query=SELECT+*+FROM+users+WHERE+id=?&outputType=SelectOne")

// INSERT/UPDATE/DELETE -> Out.Body = lignes affectées (int64)
builder.From("direct:insert").
    SetHeader(gocamel.SqlParameters, []any{"alice", "alice@example.com"}).
    To("sql://appdb?query=INSERT+INTO+users(name,email)+VALUES(?,?)")

Format d’URI :

sql://<datasourceName>?query=<SQL>
sql://logical?dataSourceRef=<datasourceName>&query=<SQL>

Paramètres de requête :

Fournis via le header CamelSqlParameters ou le corps comme []any.

// Requête fixée par la configuration (défaut, recommandé)
builder.To("sql://appdb?query=SELECT+*+FROM+users+WHERE+id=?")

// Requête fournie par message — uniquement avec des headers de confiance
builder.To("sql://appdb?allowHeaderOverride=true&query=SELECT+1")

Headers de sortie :

• CamelSqlRowCount - Lignes retournées/affectées
• CamelSqlColumnNames - Noms des colonnes (SELECT)

Corps du résultat :

Classification des requêtes :

Le choix entre exécution en lecture (jeu de résultats) et en écriture (lignes affectées) se fait sur le premier mot-clé, en ignorant les commentaires et parenthèses de tête. SELECT, WITH (CTE), VALUES, TABLE, SHOW, EXPLAIN, DESCRIBE et PRAGMA renvoient des lignes, tout comme une requête portant une clause RETURNING :

WITH recent AS (SELECT * FROM orders WHERE ts > ?)
SELECT id, total FROM recent          -- jeu de résultats
INSERT INTO users(name) VALUES (?) RETURNING id   -- jeu de résultats

SQL-Stored

Exécution de procédures stockées avec support des paramètres IN, OUT et INOUT.

import (
    "database/sql"
    _ "modernc.org/sqlite"
)

db, _ := sql.Open("mysql", "user:pass@/mydb")

sqlStored := gocamel.NewSQLStoredComponent()
sqlStored.RegisterDataSource("mydb", db)
ctx.AddComponent("sql-stored", sqlStored)

// Appel simple avec paramètres IN
builder.From("direct:call").
    SetBody([]gocamel.StoredProcedureParam{
        {Name: "userId", Direction: gocamel.ParamDirectionIn, Value: 42},
    }).
    To("sql-stored://mydb?procedure=GET_USER_BY_ID").
    Log("${body}")

Format d’URI :

sql-stored://datasourceName?procedure=NAME
sql-stored://logical?dataSourceRef=dsName&procedure=NAME

Directions des paramètres :

Exemple :

params := []gocamel.StoredProcedureParam{
    {Name: "inParam", Direction: gocamel.ParamDirectionIn, Value: "input"},
    {Name: "outParam", Direction: gocamel.ParamDirectionOut},
    {Name: "inOutParam", Direction: gocamel.ParamDirectionInOut, Value: 123},
}

MongoDB

Intégration MongoDB pour les opérations CRUD. Producer uniquement.

Format d’URI

mongodb://connectionName?database=mydb&collection=mycoll&operation=find

Options

Par sécurité, remove refuse les filtres nuls ou vides. Définissez allowDeleteAll=true explicitement uniquement si la suppression de toute la collection est intentionnelle. De même, update refuse un filtre absent ou vide (nil, {}, bson.M / bson.D vides, chaîne vide) : chacun mettrait à jour tous les documents. Définissez allowUpdateAll=true explicitement uniquement si c’est intentionnel.

⚠️ Sécurité : la base, la collection et l’opération configurées dans l’URI constituent une configuration de confiance. Les headers CamelMongoDbDatabase, CamelMongoDbCollection, CamelMongoDbOperation et CamelMongoDbCriteria sont ignorés sauf si l’endpoint active explicitement allowHeaderOverride=true. Ne l’activez que si les headers ne peuvent pas être influencés par une entrée non fiable.

Headers d’entrée

Headers de sortie

Exemple

import (
    "context"
    "go.mongodb.org/mongo-driver/mongo"
    "go.mongodb.org/mongo-driver/mongo/options"
)

// Créer le composant
mongoComp := gocamel.NewMongoDBComponent()

// Se connecter
client, _ := mongo.Connect(context.Background(), options.Client().ApplyURI("mongodb://localhost:27017"))
conn := gocamel.CreateMongoDBConnection(client, "mongodb://localhost:27017", "mydb")
mongoComp.RegisterConnection("myconn", conn)

builder.WithComponent("mongodb", mongoComp)

// Insertion
builder.From("timer:tick?period=5s").
    SetBody(map[string]any{"name": "test", "value": 123}).
    To("mongodb://myconn?database=mydb&collection=items&operation=insert")

// Requête avec filtre
builder.From("direct:search").
    SetHeader(gocamel.CamelMongoDbCriteria, map[string]any{"status": "active"}).
    SetHeader(gocamel.CamelMongoDbLimit, 10).
    To("mongodb://myconn?database=mydb&collection=items&operation=find")

Composants de Transformation

XSLT

Transformation XML via feuille de style XSL.

builder.To("xslt:file://transform.xsl")

XSD

Validation de schéma XML.

builder.To("xsd:file://schema.xsd")

Template

Traitement de templates Go (inspiré d’Apache Camel Velocity).

// Template de base
builder.To("template:templates/email.tmpl")

// Avec cache
builder.To("template:templates/item.tmpl?contentCache=true")

// Template dynamique depuis un header
builder.To("template:default.tmpl?allowTemplateFromHeader=true")

Variables de template :

{{.Body}}              # Corps du message
{{.Headers.name}}      # Valeur de header
{{.Exchange.ID}}       # ID de l'Exchange
{{.Exchange.Created}}  # Horodatage de création

Fonctions de template :

{{.Body | upper}}
{{.Body | lower}}
{{.Body | trim}}
{{now | formatDate "2006-01-02 15:04:05"}}
{{"hello" | contains "ell"}}

Composants d’Exécution

Exec

Exécuter des commandes système.

builder.To("exec:ls -la")

Surcharges par message (opt-in) :

Quand allowHeaderOverride=true, ces headers surchargent les paramètres URI pour un seul message :

Sortie :

• Out.Body = stdout de la commande (également recopié sur In pour la rétrocompatibilité, afin qu’une route InOut voie la sortie de la commande)
• Header CamelExecExitValue = code de sortie
• Headers CamelExecStdout / CamelExecStderr = sortie brute de la commande

Configuration des Composants

Authentification

Les identifiants peuvent être fournis via des variables d’environnement :

// FTP avec variable d'environnement
builder.From("ftp://host?username=${env:FTP_USER}")

// Ou paramètre
builder.From("ftp://host?username=admin&password=${env:FTP_PASS}")

Options communes

De nombreux composants partagent des options d’interrogation communes :

// Interrogation de fichiers
builder.From("file://data?delay=10s&delete=true")

// Interrogation FTP
builder.From("ftp://host/incoming?delay=30s&include=*.xml")

Tableau Récapitulatif des Composants

Détails de livraison et de cycle de vie

• Kafka : le topic résolu est porté par chaque message ; les topics configurés et surchargés par en-tête fonctionnent ensemble. Un traitement en échec est réessayé avec un nouvel échange avant la lecture suivante. Un commit en échec est réessayé sans refaire le traitement réussi. Les tentatives sont espacées de 100 ms et cessent à l’annulation. Un échec permanent bloque ce consommateur jusqu’à sa résolution par la route ou son arrêt, afin de préserver les offsets cumulatifs.
• FTP : chaque producteur réserve sa connexion exclusivement pendant tout le transfert. Les envois concurrents et Stop ne peuvent pas entrelacer les commandes FTP ; un envoi en attente de réservation peut être annulé.
• MongoDB : save préserve le _id du document entrant, même si le remplacement échoue ; le retry cible ainsi le même document.
• Mail : IMAP IDLE réagit aux notifications de boîte en terminant IDLE, en recherchant les messages puis en reprenant IDLE. Une première recherche traite les messages déjà présents.
• Cron : Stop empêche de nouvelles exécutions et attend les jobs actifs avant de retourner, y compris lors d’une pause.
• Redis : les producteurs et consommateurs actifs partagent leurs clients, fermés au départ du dernier utilisateur. Arrêter un producteur laisse les autres utilisables ; les consommateurs arrêtés libèrent leur enregistrement pour permettre le redémarrage de la route.
• File : une URI terminée par / désigne un répertoire même s’il n’existe pas encore ; CamelFileName sélectionne le fichier à l’intérieur.
• Net : le timeout configuré borne les écritures établies ainsi que la lecture des réponses. Annuler l’échange ferme sa connexion et interrompt ces opérations.
• Prometheus : des appels répétés à Metrics() ou MetricsWith(registry) réutilisent les collecteurs compatibles déjà enregistrés, ce qui permet des intercepteurs distincts sur plusieurs routes. Les définitions incompatibles échouent toujours à l’enregistrement.