📘 xcraft-core-transport

Aperçu

Le module xcraft-core-transport fournit les backends de transport pour les modules [xcraft-core-bus] et [xcraft-core-busclient]. Il implémente une couche d'abstraction pour la communication inter-processus et réseau dans l'écosystème Xcraft, supportant plusieurs protocoles de transport (EventEmitter local, Axon TCP/TLS/Unix sockets) avec des fonctionnalités avancées comme le streaming de données, le routage intelligent et la gestion des certificats TLS.

Structure du module

Le module s'organise autour de plusieurs composants principaux :

Router (lib/router.js) : Gestionnaire central de routage et d'orchestration des communications, maintient la table ARP et les lignes de connexion.
Backends (lib/backends/) : Implémentations spécifiques des protocoles de transport — axon.js pour TCP/TLS/Unix sockets et ee.js pour EventEmitter local.
Cache (lib/cache.js) : Système de mise en cache optimisé pour les expressions régulières de routage.
Streamer (lib/streamer.js) : Gestionnaire de flux de données pour le transfert de fichiers et streams volumineux.
Helpers (lib/helpers.js) : Utilitaires de sérialisation, désérialisation et manipulation des messages Xcraft.
transport.js : Exposition des commandes Xcraft pour l'administration et le monitoring.

Fonctionnement global

Le transport Xcraft fonctionne selon une architecture modulaire avec plusieurs couches :

Architecture de routage

Le système utilise une table ARP (Address Resolution Protocol) pour maintenir les routes vers les différents orcs (acteurs) du système. Chaque orc est identifié par un nom unique (orcName) et peut être accessible via les backends axon ou ee. La table ARP stocke pour chaque orc : son identifiant, token, socket, port, liste de hordes et indicateurs de configuration.

arp = {
  ee: { [orcName]: { id, token, socket, port, hordes, noForwarding, nodeName } },
  axon: { [orcName]: { id, token, socket, port, hordes, noForwarding, nodeName } }
}

Modes de communication

Push/Pull : Communication point-à-point pour les commandes (le client push envoie au serveur pull).
Pub/Sub : Communication broadcast ou routée pour les événements (le serveur pub diffuse aux clients sub).
Streaming : Transfert de fichiers et données volumineuses via le Streamer.

Gestion des lignes (Lines)

Le système maintient des "lignes" représentant les connexions logiques entre acteurs, réparties en trois catégories :

local : Lignes locales (greathall::) gérées en mémoire avec Immutable.js.
remotes : Lignes distantes synchronisées depuis le warehouse.
pending : Lignes en attente de mise à jour — les messages à destination de ces lignes sont mis en queue jusqu'à réception des nouvelles tables.

Un numéro de generation permet d'éviter les mises à jour obsolètes.

Flux de routage des messages (pub)

send(topic, msg)
  └─ _pub(topic, msg)
       ├─ Forwarding appId ? → chercher dans ARP par hordes
       ├─ lineId présent ? → résoudre via lines.local / lines.remotes
       │    └─ pending ? → mettre en queue
       ├─ arpEntries trouvées → sendTo(port, backend)
       └─ aucune entrée → broadcast sur tous les backends

Streaming de données

Le Streamer gère les transferts en deux modes :

Envoi : souscrit à stream.started.<streamId> puis pipe le stream source par chunks via emit-chunk.
Réception : souscrit à stream.chunked.<streamId> et stream.ended.<streamId>, écrit dans le stream destination.

Un timeout de 120 secondes est appliqué côté réception et se rafraîchit à chaque chunk reçu.

Exemples d'utilisation

Configuration d'un routeur push/pull (commandes)

const {Router} = require('xcraft-core-transport');
const xLog = require('xcraft-core-log')('transport', null);

// Serveur de commandes
const server = new Router('server-1', 'pull', xLog);
server.on('message', (topic, msg) => {
  console.log(`Commande reçue: ${topic}`, msg);
});
server.start({host: '127.0.0.1', port: 3334}, () => {
  console.log('Serveur démarré');
});

// Client de commandes
const client = new Router('client-1', 'push', xLog);
client.connect('axon', {host: '127.0.0.1', port: 3334}, () => {
  client.send('my-command', {_xcraftMessage: true, data: {foo: 'bar'}});
});

Configuration d'un routeur pub/sub (événements)

const {Router} = require('xcraft-core-transport');
const xLog = require('xcraft-core-log')('transport', null);

// Serveur d'événements
const pubRouter = new Router('server-1', 'pub', xLog);
pubRouter.start({host: '127.0.0.1', port: 3335}, () => {
  pubRouter.send('my-actor::event-name', {_xcraftMessage: true, data: {}});
});

// Client d'événements
const subRouter = new Router('client-1', 'sub', xLog);
subRouter.subscribe('my-actor::*');
subRouter.on('message', (topic, msg) => {
  console.log(`Événement: ${topic}`, msg);
});
subRouter.connect('axon', {host: '127.0.0.1', port: 3335}, () => {});

Utilisation des commandes de transport (monitoring)

// Affichage de la table ARP via le bus Xcraft
const busClient = require('xcraft-core-busclient').getGlobal();
const resp = busClient.newResponse('transport', 'token');

// Voir la table ARP
resp.command.send(`${appId}.arp`, {}, (err) => {
  // La table ARP est affichée dans les logs
});

// Voir le statut de tous les routeurs
resp.command.send(`${appId}.status`, {}, (err, status) => {
  console.log('Statut:', status);
});

// Extraire les métriques Xcraft
resp.command.send('transport.xcraftMetrics', {}, (err, metrics) => {
  console.log('Métriques:', metrics);
});

Connexion avec TLS

const client = new Router('client-1', 'push', xLog);
client.connect(
  'axon',
  {
    host: 'remote-server.example.com',
    port: 3334,
    caPath: '/path/to/ca.pem', // Active TLS
    keyPath: '/path/to/client.key', // Optionnel: certificat client
    certPath: '/path/to/client.pem',
  },
  () => {
    console.log('Connecté via TLS');
  }
);

Interactions avec d'autres modules

Le module xcraft-core-transport interagit étroitement avec :

xcraft-core-bus : Fournit l'infrastructure de transport pour le bus de commandes ; le Router utilise xcraft-core-bus pour récupérer le token d'authentification et publier les événements de routage.
xcraft-core-busclient : Utilise les transports pour les connexions client ; le Streamer et le Router l'utilisent pour envoyer des commandes et événements de streaming.
xcraft-core-etc : Charge la configuration du module (xcraft-core-transport) pour déterminer les backends activés et les options TLS.
xcraft-core-host : Fournit appId, appMasterId et getRoutingKey() pour le routage applicatif et la gestion des hordes.
xcraft-core-horde : Intégration optionnelle pour gérer les connexions multi-instances (slaves) dans la table ARP.
xcraft-core-probe : Collecte les métriques de performance des opérations de transport (push, pub, subscribe).
xcraft-core-utils : Utilitaires regex pour la conversion des patterns de topics en expressions régulières.
xcraft-core-shredder : Utilisé par les helpers pour sérialiser/désérialiser les payloads de type Shredder.

Configuration avancée

Option	Description	Type	Valeur par défaut
`backends`	Liste des backends activés (vide = tous)	Array	`[]`
`axon.clientOnly`	Mode client uniquement pour Axon (désactive l'écoute serveur)	Boolean	`false`
`requestClientCert`	Demander les certificats clients TLS	Boolean	`false`
`certsPath`	Emplacement des certificats clients	String	`{xcraftRoot}/var/certs`
`keysPath`	Emplacement des clés privées clients	String	`{xcraftRoot}/var/keys`

Variables d'environnement

Ce module n'utilise pas de variables d'environnement directement. La configuration passe par [xcraft-core-etc] et [xcraft-core-host].

Détails des sources

`transport.js`

Expose les commandes Xcraft sur le bus pour l'administration et le monitoring des transports. Le namespace des commandes est construit dynamiquement à partir de appId et du tribe courant (ex: myapp.arp, myapp-tribe1.status).

Commandes publiques

{appId}.arp — Affiche la table ARP complète (backend, orcName, id, token, port, hordes) dans les logs.
{appId}.arp.hordes — Affiche la répartition des hordes dans la table ARP avec représentation visuelle des longueurs.
{appId}.lines — Affiche les lignes de connexion actives avec leurs compteurs de référence par orc.
{appId}.status — Affiche le statut détaillé de tous les routeurs (mode, backends actifs, souscriptions).
{appId}.emit-chunk — Émet un chunk de données dans le système de streaming (utilisé par le Streamer).
{appId}.emit-end — Signale la fin d'un stream.
{appId}.start-emit — Démarre l'émission d'un stream avec la routing key de destination.
xcraftMetrics — Extrait les métriques Xcraft pour le monitoring (taille ARP, lignes, sockets, bytes read/written).

`lib/index.js`

Point d'entrée principal du module. Expose les classes et utilitaires principaux, et gère un registre global (registry) des routeurs instanciés. La classe WrappedRouter étend Router pour s'auto-enregistrer dans ce registre à la construction et se désenregistrer à l'arrêt.

Exports principaux :

helpers — Instance des utilitaires de sérialisation.
Cache — Classe de cache pour les expressions régulières.
Router — Classe WrappedRouter avec gestion du registre.
getRouters() — Retourne la liste plate de tous les routeurs actifs.

`lib/router.js`

Cœur du système de routage qui orchestre la communication entre les différents backends de transport. Charge dynamiquement les backends disponibles depuis lib/backends/ en filtrant selon la configuration.

Tables de données globales

arp : Table ARP partagée entre toutes les instances de Router, indexée par backend (ee, axon) puis par orcName.
lines : Lignes logiques avec local (Immutable Map), remotes (Map par horde), pending (Map lineId→generation) et generation courant.
routers : Registre des routeurs par id et mode, partagé avec WrappedRouter.

Méthodes publiques d'instance

start(options, callback) — Démarre le routeur sur tous les backends configurés. Accepte host, port, unixSocketId, timeout, keyPath, certPath, serverKeepAlive.
stop() — Arrête proprement le routeur et ferme toutes les connexions backend.
connect(backend, options, callback) — Connecte le routeur à un backend spécifique en mode client (push ou sub).
send(topic, ...args) — Envoie un message via le routeur (délègue à _push en mode connecté, ou _pub en mode serveur).
subscribe(topic, backend, orcName) — Souscrit à un topic avec gestion automatique des lignes. Retourne {ids, str, reg}.
unsubscribe(topic, backend, orcName) — Se désinscrit d'un topic et met à jour les lignes si nécessaire.
on(topic, handler, proxy=false) — Écoute les événements sur tous les backends.
hook(topic, handler) — Enregistre un handler pour un topic interne du routeur (message, error, disconnect, etc.).
status() — Retourne l'état courant : backends, options, mode, connectedWith.
acceptIncoming() — Autorise les nouvelles connexions entrantes (utile après une phase d'initialisation).
onInsertOrc(handler) — Écoute l'insertion d'un nouvel orc dans la table ARP.
onDeleteOrc(handler) — Écoute la suppression d'un orc de la table ARP.
connectedWith() — Retourne le nom du backend actuellement utilisé pour la connexion client.
destroySockets() — Détruit tous les sockets ouverts sur tous les backends.

Méthodes statiques

deleteRoute(orcName, backend) — Supprime une route de la table ARP (et supprime aussi l'entrée ee si backend est axon).
updateLines(_lines, _token, generation, horde) — Met à jour les lignes distantes et traite les messages en queue pour les lignes qui n'étaient plus en attente.
connectLine(lineId, orcName) — Établit une connexion logique (locale ou via warehouse.request-line-update).
disconnectLine(lineId, orcName) — Ferme une connexion logique.
moveRoute(oldOrcName, newOrcName) — Déplace une route lors de l'autoconnect (renommage d'orc temporaire → définitif).
getRoute(orcName, backend) — Récupère une entrée ARP.
getRouters(orcName, backend) — Retourne le groupe de routeurs associé à une route.
getARP() — Retourne la table ARP globale.
getLines() — Retourne l'objet lines global.
getNice(orcName, backend) — Retourne la priorité (nice) d'un orc.
extractLineId(topic) — Extrait l'ID de ligne d'un topic (délègue aux helpers).

`lib/cache.js`

Système de cache optimisé pour les expressions régulières utilisées dans le routage des messages. Au lieu de tester toutes les regex sur chaque topic, le cache extrait d'abord les IDs présents dans le topic (via extractIds) pour ne tester que les regex associées à ces IDs spécifiques, réduisant drastiquement le nombre de comparaisons.

Méthodes publiques

matches(topic) — Vérifie si le topic correspond à au moins une regex en cache. Retourne true si une correspondance est trouvée.
map(topic, predicate) — Collecte et retourne les valeurs mappées par le prédicat pour toutes les regex qui correspondent au topic.
set(id, key, value) — Ajoute une regex (value) au cache sous la clé id (extrait du topic) et key (généralement regex.toString()).
del(id, key) — Supprime une entrée du cache. Si l'ID n'a plus d'entrées, il est supprimé.
clear() — Vide complètement le cache.

`lib/helpers.js`

Utilitaires pour la sérialisation/désérialisation des messages Xcraft avec support des types spéciaux (Immutable.js, Shredder), la gestion des streams et l'extraction d'identifiants depuis les topics.

Le cache interne idsCache limite sa taille à 4096 entrées (suppression par lot de 128) pour éviter les fuites mémoire.

Fonctions principales

extractIds(topic) — Extrait et ordonne les identifiants d'un topic (format user@domain, UUID, <lineId>) pour optimiser les lookups dans le Cache. Résultat mis en cache.
extractLineId(topic) — Extrait l'ID de ligne au format <lineId> d'un topic.
toXcraftJSON(args, newStreamer) — Sérialise les arguments en format Xcraft JSON : détecte les streams à transférer, encode les types Immutable (_xImmu) et Shredder (_xShred), et sépare les chunks buffers en arguments distincts pour optimiser la sérialisation Axon.
fromXcraftJSON(args, newStreamer) — Désérialise depuis le format Xcraft JSON en restaurant les types Immutable/Shredder et en reconstituant les chunks buffers.
dataToXcraftJSON(d) — Convertit récursivement les données Immutable/Shredder en représentation JSON sérialisée avec marqueurs de type.
dataFromXcraftJSON(d, root=false) — Restaure récursivement les données depuis leur représentation JSON Xcraft.
tryStreamTo(msg, newStreamer) — Détecte et initialise le streaming pour un message contenant xcraftUpload ou xcraftStream.
restoreChunkBuffer(args) — Reconstitue le buffer de chunk séparé dans les arguments de message.

`lib/streamer.js`

Gestionnaire de flux de données pour le transfert de fichiers et streams volumineux à travers le réseau Xcraft. Chaque instance gère un seul transfert identifié par un streamId.

Le Streamer fonctionne selon deux directions :

Envoi (constructeur appelé avec stream) : attend l'événement stream.started.<streamId> puis pipe le stream source par chunks.
Réception (constructeur appelé sans stream) : souscrit aux événements stream.chunked.<streamId> et stream.ended.<streamId>.

Le mécanisme de streaming utilise des commandes RPC (transport.{routingKey}.emit-chunk, emit-end, start-emit) lorsque l'émetteur et le récepteur sont sur des nœuds différents. Pour les transferts locaux (routingKey identique), les événements Xcraft sont utilisés directement.

Un contournement pour un bug AMP (s:, b:, j: au début des buffers) est appliqué avant l'envoi de chaque chunk.

Méthodes publiques

receive(remoteRoutingKey, stream, progress, next) — Lance la réception d'un stream vers stream (Writable). Appelle progress(current, total) à chaque chunk et next(err) à la fin ou en cas d'erreur. Timeout de 120s.

`lib/backends/axon.js`

Backend de transport basé sur la bibliothèque xcraft-axon, supportant TCP, TLS et Unix sockets pour la communication réseau inter-processus ou réseau.

Fonctionnalités avancées

TLS : Support complet avec certificats serveur (keyPath, certPath) et client optionnel (requestClientCert). Les certificats statiques sont chargés depuis resources/certs/ et les certificats dynamiques depuis certsPath. La surveillance des fichiers avec chokidar permet le rechargement à chaud via refreshCerts().
Unix sockets : Communication haute performance sur Linux via des sockets Unix (identifiés par inodes via ss et /proc). Désactivé sur Windows.
Gestion des inodes : Pour les Unix sockets, l'inode côté serveur est résolu pour permettre le routage par "port" (qui est en fait l'inode).
Retry automatique : En cas de conflit de port (EADDRINUSE, EACCES), le port est incrémenté et une nouvelle tentative est effectuée.
Queue de connexion : Les messages autoconnect peuvent être mis en attente si acceptIncoming n'est pas encore activé.

Méthodes publiques

connect(options, callback) — Établit une connexion client. Options : host, port, timeout, caPath (TLS), keyPath, certPath, clientKeepAlive, unixSocketId.
start(options, callback) — Démarre le serveur. Options : host, port, timeout, keyPath, certPath, serverKeepAlive, unixSocketId.
stop() — Ferme proprement le socket et le watcher TLS.
send(topic, streamChannel, ...args) — Envoie un message (broadcast à tous les sockets connectés).
sendTo(port, topic, streamChannel, ...args) — Envoie vers un socket spécifique identifié par son port (TCP) ou inode (Unix).
subscribe(re, ids) — Souscrit à des patterns de messages (délègue au socket Axon).
unsubscribe(re) — Se désinscrit d'un pattern.
unsubscribeAll() — Supprime toutes les souscriptions.
destroySockets(ports=[]) — Détruit les sockets (tous si ports vide, sinon ceux correspondant aux ports listés).
refreshCerts() — Recharge dynamiquement les certificats TLS (certificats statiques + dynamiques depuis certsPath).
acceptIncoming() — Traite les connexions autoconnect mises en attente.
status() — Retourne {host, port, active, subscriptions}.

`lib/backends/ee.js`

Backend de transport basé sur EventEmitter Node.js pour les communications locales haute performance au sein du même processus. Les instances partagent un EventEmitter commun (ee global) identifié par host:port, ce qui permet une communication directe sans sérialisation réseau.

Chaque instance EE possède un identifiant UUID unique (_id) pour gérer le compteur de références et les événements close par instance.

Caractéristiques

Communication synchrone ultra-rapide (même processus).
Pas de sérialisation réseau — les objets sont passés par référence.
Mise en cache des souscriptions via la classe Cache pour le filtrage efficace des topics.
Support des modes pubsub et pushpull avec le même EventEmitter sous-jacent.
Gestion des onPending : les handlers enregistrés avant la connexion sont appliqués lors de connect/start.

Méthodes publiques

connect(options, callback) — Connecte à l'EventEmitter partagé identifié par host:port.
start(options, callback) — Démarre le backend EventEmitter (idempotent si déjà démarré).
stop() — Émet l'événement close et supprime tous les handlers.
send(topic, streamChannel, ...args) — Émet un message local (appel synchrone).
sendTo(port, topic, streamChannel, msg, ...args) — Identique à send (le port est ignoré pour EE).
subscribe(re, ids) — Souscrit à un pattern de messages et l'ajoute au Cache.
unsubscribe(re) — Se désinscrit d'un pattern.
unsubscribeAll() — Supprime toutes les souscriptions.
fixId(oId, nId) — Renomme la clé de l'EventEmitter partagé (utilisé lors des changements d'adresse).
status() — Retourne {active, subscriptions}.

Licence

Ce module est distribué sous licence MIT.

Ce contenu a été généré par IA

Name		Name	Last commit message	Last commit date
Latest commit History 554 Commits
lib		lib
test		test
.editorconfig		.editorconfig
.gitignore		.gitignore
.zou-flow		.zou-flow
LICENSE		LICENSE
README.md		README.md
config.js		config.js
eslint.config.js		eslint.config.js
package.json		package.json
transport.js		transport.js

Folders and files

Latest commit

History

Repository files navigation

📘 xcraft-core-transport

Aperçu

Sommaire

Structure du module

Fonctionnement global

Architecture de routage

Modes de communication

Gestion des lignes (Lines)

Flux de routage des messages (pub)

Streaming de données

Exemples d'utilisation

Configuration d'un routeur push/pull (commandes)

Configuration d'un routeur pub/sub (événements)

Utilisation des commandes de transport (monitoring)

Connexion avec TLS

Interactions avec d'autres modules

Configuration avancée

Variables d'environnement

Détails des sources

transport.js

Commandes publiques

lib/index.js

lib/router.js

Tables de données globales

Méthodes publiques d'instance

Méthodes statiques

lib/cache.js

Méthodes publiques

lib/helpers.js

Fonctions principales

lib/streamer.js

Méthodes publiques

lib/backends/axon.js

Fonctionnalités avancées

Méthodes publiques

lib/backends/ee.js

Caractéristiques

Méthodes publiques

Licence

About

Resources

License

Uh oh!

Stars

Watchers

Forks

Releases

Packages 0

Uh oh!

Uh oh!

Contributors

Uh oh!

Languages

`transport.js`

`lib/index.js`

`lib/router.js`

`lib/cache.js`

`lib/helpers.js`

`lib/streamer.js`

`lib/backends/axon.js`

`lib/backends/ee.js`

Packages