Camel Attachment et JMS

Bonjour.

Je rencontre régulièrement le même problème. je reçois d'une ressource externe un message auquel est associé une ou plusieurs pièces jointes.
tout cela transite via JMS et est distribué à un ou plusieurs destinataires.

Camel est parfait pour cela.
Je travaille sous OSGI(karaf) et le bundle qui dépose dans la file jms n'est pas le même que celui qui s'abonne à la file.
Il n'est pas nécessairement sur la même instance de karaf. de plus l'un ou l'autre peut être arrêté.

Tout ça pour dire que je ne peux me passer du broker.
pour résoudre le problème des pièce jointe à chaque fois je me suis défini un format de message ad-hoc sérialisable contenant le tout.

J'en suis venu à essayer de trouver une solution commune objet de ce post.

partons d'un exemple simple (enfin pas trop)
une route camel lit des fichiers XML dans un dossier
elle utilise un bean pour transformer le contenu, lire une pièce jointe, et garder aussi le message original (un besoin récurant dans mon secteur)
puis elle place tout ça dans une file JMS.

une deuxième route lit la file
vérifie la présence du message original
et produit deux fichier dans un répertoire.

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
    from("file://work/jms/input?preMove=../inprogress/&sortBy=file:name&move=../in/&include=NG.*xml")
    .routeId("file2jms")
    .convertBodyTo(String.class)
    .to("log:In")
    .bean(SampleConverter.class, "convertFrom")
    .to("activemq:queue:LOG.ME")
    ;
 
    from("activemq:queue:LOG.ME")
    .routeId("jms2file")
    .bean(SampleConverter.class, "convertTo")
    .to("log:Out")
    //produire un fichier pour le pdf et un pour le body
    .split().method(SampleSplitter.class, "split")
    .to("file://work/jms/output")
    ;

le SampleConverter fait les opérations décrite ci dessus
le SampleSplitter sépare le body et la pièce jointe en deux message pour les déposer dans le dossier.
il s'agit d'un exemple et il n'est pas du tout adapté pour garantir que les éléments arriverons tous sans être séparé.
Il est juste là pour avoir le problème d'un body et plusieurs attachement.

Code:

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
  public void convertFrom(Exchange exchange) throws Exception {
    Logger log = LoggerFactory.getLogger(SampleConverter.class);
 
    Message msg = exchange.getIn();
    String body = msg.getBody(String.class);
    // garder l'original en attachement
    DataHandler content = StringToDataHandler(body);
    msg.addAttachment(ORIGINAL, content);
 
    // transformer le contenu
    msg.setBody(body.replaceFirst("Test", "Test2"));
 
    // récupérer le fichier joint et le placer en attachement
    String pdfFileName = normalizePath(msg.getHeader("CamelFileAbsolutePath", String.class))
        .replaceAll("\\.xml", "\\.pdf");
    File pdfFile = new File(pdfFileName);
    if (null != pdfFile) {
      log.info("file : " + pdfFile);
      msg.addAttachment(PDF, FileToDataHandler(pdfFile));
    }
  }
 
  public void convertTo(Exchange exchange) throws Exception {
    Logger log = LoggerFactory.getLogger(SampleConverter.class);
    Message msg = exchange.getIn();
 
    // afficher le contenu de l'original dans les log
    DataHandler original = msg.getAttachments().get(ORIGINAL);
    log.info("original : " + dataHandlerToString(original));
    if (null != msg.getAttachments().get(PDF)) {
      log.info("pdf : ok");
    }
  }

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
  public List<Message> split(Message msg) throws IOException {
    List<Message> answer = new ArrayList<Message>();
 
    // si on a une pièce jointe nommée pdf l'ajouter à la liste des fichiers
    // à produire
    DataHandler pdf = msg.getAttachments().get(PDF);
    if (null != pdf) {
      Message message = msg.copy();
      message.setBody(dataHandlerToBytes(pdf));
      message.setHeader(FILE_NAME, message.getHeader(FILE_NAME, String.class).replaceAll("\\.xml", "\\.pdf"));
      answer.add(message);
    }
    // ajouter le message courant à la liste des fichiers à produire
    answer.add(msg);
    return answer;
  }

Le problème à résoudre est comment faire passer le message au travers de JMS.
la première chose est que le contenu du message doit être sérialisable.
Les headers définis dans Camel sont préservé ainsi que le contenu.
Mais les attachement sont perdus.

à y regarder de plus près on constate qu'on est un peut dans le même cas qu'un mail.
pour résoudre ce problème (à une époque ou XML n'existait pas) on a défini MIME multiPart

nous allons faire de même mais en XML

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
<?xml version="1.0" encoding="ISO-8859-1"?>
<xsd:schema targetNamespace="urn:org.apache.camel.envelope"
 xmlns="urn:org.apache.camel.envelope" xmlns:xsd="http://www.w3.org/2001/XMLSchema"
 xmlns:xmime="http://www.w3.org/2005/05/xmlmime"
 elementFormDefault="qualified" attributeFormDefault="unqualified">
    <xsd:complexType name="Envelope">
        <xsd:sequence>
            <xsd:element name="body" type="xsd:base64Binary"/>
            <xsd:element maxOccurs="unbounded" minOccurs="0" name="part" type="Part"> </xsd:element>
        </xsd:sequence>
    </xsd:complexType>
    <xsd:complexType name="Part">
        <xsd:simpleContent>
            <xsd:extension base="xsd:base64Binary"
                xmime:expectedContentTypes="application/octet-stream">
                <xsd:attribute name="name" type="xsd:token"/>
            </xsd:extension>
        </xsd:simpleContent>
    </xsd:complexType>
    <xsd:element name="message" type="Envelope"/>
</xsd:schema>

Un body est autant de parts que l'on désire.
Reste à transformer le message avec des attachement en envelope et à la sérialiser en xml.
puis faire l'opération inverse.

un peu de jaxb et camel nous offre une api pour définir notre propre transformateur.

Code:

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
public class EnvelopeDataFormat implements DataFormat {
 
  private static JAXBContext JC;
  private static Unmarshaller UNMARSHALLER;
  private static Marshaller MARSHALLER;
 
  {
    try {
      JC = JAXBContext.newInstance(Envelope.class.getPackage().getName());
      UNMARSHALLER = JC.createUnmarshaller();
      MARSHALLER = JC.createMarshaller();
    } catch (Exception e) {
      Logger log = LoggerFactory.getLogger(EnvelopeDataFormat.class);
      log.error("cannot create dataformat", e);
    }
  }
 
  /**
   * encode le document et les attachements dans une envelope
   *
   * @param exchange échange Camel
   * @param graph document
   * @param stream flux de sortie
   */
  public void marshal(Exchange exchange, Object graph, OutputStream stream) throws Exception {
    String body = ExchangeHelper.convertToMandatoryType(exchange, String.class, graph);
 
    Message msg = exchange.getIn();
 
    ObjectFactory of = new ObjectFactory();
 
    Envelope envelope = of.createEnvelope();
    envelope.setBody(body.getBytes());
    Set<Entry<String, DataHandler>> attachements = msg.getAttachments().entrySet();
    for (Entry<String, DataHandler> entry : attachements) {
      Part part = of.createPart();
      part.setName(entry.getKey());
      part.setValue(entry.getValue());
      envelope.getPart().add(part);
    }
 
    MARSHALLER.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, Boolean.TRUE);
    MARSHALLER.marshal(of.createMessage(envelope), stream);
  }
 
  /**
   * décode un message Envelope
   *
   * @param exchange échange camel.
   * @param stream flux contenant l'envelope
   * @return le document nouvellement créé.
   */
  public Object unmarshal(Exchange exchange, InputStream inputStream) throws Exception {
    StreamSource stream = new StreamSource(inputStream);
    Envelope envelope = UNMARSHALLER.unmarshal(stream, Envelope.class).getValue();
    Message message = exchange.getIn();
 
    List<Part> parts = envelope.getPart();
    for (Part part : parts) {
      message.addAttachment(part.getName(), part.getValue());
    }
    message.setBody(new String(envelope.getBody()));
    exchange.setOut(message);
    return message.getBody(String.class);
  }
 
}

Envelope est le résultat de la génération du code java par jaxb.
pour faire un dataformat il suffit d'implémenter l'interface du même nom.
un code relativement simple.

Reste à faire en sorte que les route l'utilise.
là encore Camel nous aide
Dans le route builder ajouter

Code:

1
2
3
    //utiliser une envelope pour les échanges activemq
    interceptFrom("activemq:*").unmarshal("envelope");
    interceptSendToEndpoint("activemq:*").marshal("envelope");

pour tout les échanges de ce contexte camel via activemq les messages seront mis sous enveloppe avec leurs attachements.

les attachements camel utilise comme pour les mails des DataHandler pour facilité leur utilisation je me suis ajouté quelque méthodes

Code:

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
  private static final String TEXT_PLAIN_CHARSET_UTF_8 = "text/plain; charset=UTF-8";
 
  public static String dataHandlerToString(DataHandler dataHandler) throws IOException {
    ByteArrayOutputStream bos = new ByteArrayOutputStream();
    dataHandler.writeTo(bos);
    bos.flush();
    bos.close();
    return bos.toString();
  }
 
  public static DataHandler StringToDataHandler(String string) {
    return BytesToDataHandler(string.getBytes());
  }
 
  public static byte[] dataHandlerToBytes(DataHandler dataHandler) throws IOException {
    ByteArrayOutputStream bos = new ByteArrayOutputStream();
    dataHandler.writeTo(bos);
    bos.flush();
    bos.close();
    return bos.toByteArray();
  }
 
  public static DataHandler BytesToDataHandler(byte[] bytes) {
    return new DataHandler(new ByteArrayDataSource(bytes, TEXT_PLAIN_CHARSET_UTF_8));
  }
 
  public static DataHandler FileToDataHandler(File file) {
    return new DataHandler(new FileDataSource(file));
  }

A+JYT