Keine Nachrichten Wenn Input Stream von SFTP Outbound-Gateway Abrufen

Question

Keine Nachrichten Manchmal Wenn Input Stream von SFTP Outbound Gewinnung Gateway-

Dies ist Follow-up-Frage zu Use SFTP Outbound Gateway to Obtain Input Stream

Das Problem, das ich in vorheriger Frage haben scheint, dass Ich habe den Stream nicht wie im Dienst int: service-activator geschlossen. Als ich jedoch den int: service-activator hinzugefügt habe, schien ich gezwungen zu sein, int: poller hinzuzufügen.

Wenn ich jedoch den int: Poller hinzugefügt habe, habe ich bemerkt, dass die Nachrichten beim Versuch, den Stream zu erhalten, manchmal null sind. Ich habe festgestellt, dass eine Problemumgehung ist, einfach erneut zu versuchen. Ich habe mit verschiedenen Dateien getestet und es scheint, dass kleine Dateien nachteilig betroffen sind und große Dateien nicht. Also, wenn ich raten musste, muss es eine Race-Bedingung geben, bei der der Int: Service-Aktivator die Sitzung schließt, bevor ich getInputStream() anrufe, aber ich hatte gehofft, dass jemand erklären könnte, ob das tatsächlich passiert und ob es da ist ist eine bessere Lösung als einfach nur versuchen?

Danke!

ist die Ausgangs-Gateway-Konfiguration:

<int-ftp:outbound-gateway session-factory="ftpClientFactory" 
      request-channel="inboundGetStream" command="get" command-options="-stream" 
      expression="payload" remote-directory="/" reply-channel="stream"> 
    </int-ftp:outbound-gateway> 

    <int:channel id="stream"> 
      <int:queue/> 
    </int:channel> 

    <int:poller default="true" fixed-rate="50" /> 

    <int:service-activator input-channel="stream" 
       expression="payload.toString().equals('END') ? headers['file_remoteSession'].close() : null" /> 

Hier ist die Quelle, wo ich die Input erhalten:

public InputStream openFileStream(final int retryCount, final String filename, final String directory) 
       throws Exception { 
      InputStream is = null; 
      for (int i = 1; i <= retryCount; ++i) { 
       if (inboundGetStream.send(MessageBuilder.withPayload(directory + "/" + filename).build(), ftpTimeout)) { 
         is = getInputStream(); 
         if (is != null) { 
           break; 
         } else { 
           logger.info("Failed to obtain input stream so attempting retry " + i + " of " + retryCount); 
           Thread.sleep(ftpTimeout); 
         } 
       } 
      } 
      return is; 
    } 

    private InputStream getInputStream() { 

      Message<?> msgs = stream.receive(ftpTimeout); 

      if (msgs == null) { 
       return null; 
      } 

      InputStream is = (InputStream) msgs.getPayload(); 
      return is; 
    } 

aktualisieren, werde ich gehen Sie vor und nehmen die einzige Antwort, wie es geholfen Gerade genug, um die Lösung zu finden.

Die Antwort auf die ursprüngliche Frage angenommene Antwort war verwirrend, weil sie eine Java-Frage mit einer xml-Konfigurationslösung beantwortete, die, während das Problem erklärt wurde, nicht wirklich die notwendige Java technische Lösung zur Verfügung stellte. Diese Follow-up-Frage/Antwort verdeutlicht, was in der Frühjahrsintegration vor sich geht und schlägt vor, was zu lösen ist.

Endgültige Lösung. Um den Stream für später zu erhalten und zu speichern, musste ich eine Bean erstellen, um den Stream zur späteren Referenz zu speichern. Dieser Stream wird aus dem Nachrichtenheader erhalten.

Hinweis, Fehlerprüfung und Getter/Setter wird der Kürze halber weggelassen:

1. Verwenden Sie die gleiche XML-Konfigurations wie in der Frage oben, aber die poller und Service-Aktivatorelemente beseitigen, da sie unnötig sind und waren die Fehler verursachen.

2. Erstellen Sie eine neue Klasse SftpStreamSession notwendigen Referenzen zu halten:

   public class SftpStreamSession { 
   
       private Session<?> session; 
       private InputStream inputStream; 
   
       public void close() { 
        inputStream.close(); 
        session.close(); 
       } 
   } 
   

3. Ändern der openFileStream Methode eine SftpStreamSession zurückzukehren:

   public SftpStreamSession openFileStream(final String filename, final String directory) throws Exception { 
   
       SftpStreamSession sss = null; 
       if (inboundGetStream.send(MessageBuilder.withPayload(directory + "/" + filename).build(), ftpTimeout)) { 
   
        Message<?> msgs = stream.receive(ftpTimeout); 
   
        InputStream is = (InputStream) msgs.getPayload(); 
   
        MessageHeaders mH = msgs.getHeaders(); 
        Session<?> session = (Session<?>) mH.get("file_remoteSession"); 
   
        sss = new SftpStreamSession(session, is); 
       } 
   
       return sss; 
   } 
   

Answer 1

allererst die Sie nicht brauchen payload.toString().equals('END') weil es so aussieht, als ob Sie <int-file:splitter> nicht in Ihrer Logik verwenden.

Sekunde. Sie brauchen diese hässliche <service-activator> nicht, weil Sie vollen Zugriff auf die Nachricht in Ihrem Code haben.Sie können einfach erhalten file_remoteSession, werfen Sie es in Session<?> und rufen Sie .close() am Ende Ihrer Logik.

Ja, es gibt eine Race-Bedingung, aber es passiert in Ihrem Code.

Schauen Sie, Sie haben streamQueueChannel. Von Anfang an hatten Sie einen Verbraucher stream.receive(ftpTimeout);. Aber jetzt haben Sie das <int:service-activator input-channel="stream"> eingeführt. Daher ein weiterer Wettbewerber Verbraucher. Wenn Sie ein so kleines (fixed-rate="50") Abfrageintervall haben, führt dies zu unerwartetem Verhalten.