Hilfe zu Design mit EJB

[mihe7]

a) Das passiert bei der Erzeugung Deines Handlers doch automatisch
b) Nein, im Idealfall rufst Du in onMailReceived die offer-Methode der Queue auf.
c) doSomethingWith(obj) ist lediglich ein Platzhalter, den Du halt mit den für Dich passenden Anweisungen ersetzen musst.

Welche Modifikation brauche ich in den oberen Klassen?

Das habe ich in #16 bereits beschrieben. Der Code aus #16 muss in Deine EJB (TicketEmailWatcherService). Bei der Erzeugung des MailPushEventHandlers musst Du die messages-Queue mitgeben (-> Anpassung des Konstruktors von MailPushEventHandler). Und in MailPushEventHandler#onMessageReceived legst Du das zu verarbeitende Objekte in die Queue mit der offer-Methode.

 

[beta20]

welche Klassen stecken denn hinter: private Queue<Object> messages; ?
Bekomme in Eclipse einen Fehler, dass er "take" nicht findet...
"The method take() is undefined for the type Queue<Object>"

 

[mrBrown]

Mach aus Queue eine BlockingQueue.

(Bei solchen Problemen hilft ein kleiner Blick ins Javadoc, im Link von @mihe7 steht drin, woher die Methode kommt )

 

[beta20]

OK, danke - klappt....
Fangen wir Schritt für Schritt an:

Passt dieser Code nun im "TicketEmailWatcherService" (EJB Klasse) ?

@Resource
    private ManagedExecutorService executor;
    private BlockingQueue<Object> messages;
    private volatile boolean running;

    /**
     * Init watching Emailaddresses
     */
    @PostConstruct
    protected void initStartTicketListener() {

        LOGGER.info("START initStartTicketListener");

        messages = new LinkedBlockingQueue<>();
        running = true;
        executor.submit(() -> {
            while (running) {

                try {
                    Object obj = messages.take();

                    List<TicketEmailSetting> list = ticketEmailSettingService
                            .findAllTicketEmailSettingListByStatusActive();
                    if (list == null || list.isEmpty())
                        return;

                    for (TicketEmailSetting ticketEmailSetting : list) {
                        Store store = connect(ticketEmailSetting.getEmailSetting());
                        MailPushEventHandler mailEventHandler = new MailPushEventHandler(store, messages);
                    }
                } catch (Exception e) {

                }
            }
        });

        LOGGER.info("END initStartTicketListener");
    }

 

[mihe7]

Passt dieser Code nun im "TicketEmailWatcherService" (EJB Klasse) ?

Nein. Der Teil

                    List<TicketEmailSetting> list = ticketEmailSettingService
                            .findAllTicketEmailSettingListByStatusActive();
                    if (list == null || list.isEmpty())
                        return;

                    for (TicketEmailSetting ticketEmailSetting : list) {
                        Store store = connect(ticketEmailSetting.getEmailSetting());
                        MailPushEventHandler mailEventHandler = new MailPushEventHandler(store, messages);
                    }

gehört nicht in den Thread. Der Code dient ja dem Start der Überwachung.

 

[beta20]

Also so???

@PostConstruct
    protected void initStartTicketListener() {

        LOGGER.info("START initStartTicketListener");

        try {

            List<TicketEmailSetting> list = ticketEmailSettingService.findAllTicketEmailSettingListByStatusActive();
            if (list == null || list.isEmpty())
                return;

            for (TicketEmailSetting ticketEmailSetting : list) {
                Store store = connect(ticketEmailSetting.getEmailSetting());
                MailPushEventHandler mailEventHandler = new MailPushEventHandler(store, messages);
            }
        } catch (Exception e) {

        }

        messages = new LinkedBlockingQueue<>();
        running = true;
        executor.submit(() -> {
            while (running) {

                try {
                    Object obj = messages.take();
                } catch (InterruptedException e) {

                }
            }
        });

        LOGGER.info("END initStartTicketListener");
    }

dann verstehe ich aber nicht, was hier passieren soll:

 executor.submit(() -> {
            while (running) {
                try {
                    Object obj = messages.take();
                } catch (InterruptedException e) {
                }
            }

 

[mihe7]

Also so???

Fast. Nimm den try-catch-Block (EDIT: der auf Ebene von initStartTicketListener) nach unten, dann ist messages auch schon initialisiert, wenn Du den Konstruktor von MailPushEventHandler aufrufst

dann verstehe ich aber nicht, was hier passieren soll:

Aktuell entnimmst Du nur das Objekt aus der Queue, machst aber damit noch nichts. Du könntest an der Stelle die Nachricht in der DB speichern, oder was auch immer Du damit anfangen willst.

 

[beta20]

Ok, try / catch is nach unten:

/**
     * Init watching Emailaddresses
     */
    @PostConstruct
    protected void initStartTicketListener() {

        LOGGER.info("START initStartTicketListener");

        messages = new LinkedBlockingQueue<>();
        running = true;

        try {

            List<TicketEmailSetting> list = ticketEmailSettingService.findAllTicketEmailSettingListByStatusActive();
            if (list == null || list.isEmpty())
                return;

            for (TicketEmailSetting ticketEmailSetting : list) {
                Store store = connect(ticketEmailSetting.getEmailSetting());
                MailPushEventHandler mailEventHandler = new MailPushEventHandler(store, messages);
            }
        } catch (Exception e) {

        }

        executor.submit(() -> {
            while (running) {

                try {
                    Object obj = messages.take();
                } catch (InterruptedException e) {

                }
            }
        });

        LOGGER.info("END initStartTicketListener");
    }

Die Klasse habe ich so angepasst:

public class MailPushEventHandler extends MailPushEmpfaenger {

    public MailPushEventHandler(Store store, Queue<Object> messagesQueue) {
        super(store);
    }

    @Override
    public void onMailReceived(Object[] messages) {

        if (messages != null && messages.length > 0) {

            for (Object msg : messages) {

                try {
                    String textFromMessage = getTextFromMessage((Message) msg);

                } catch (MessagingException e) {
                    e.printStackTrace();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

}

Im Grunde genommen bräuchte doch den Code von onMailReceived in meiner TicketEmailWatcherService - Klasse in der while - Schleife?

Mit all den "neuen" Email, die in Object[] messages sind, soll dann eben was passieren (Ticket in Datenbank anlegen).
Also eigentlich sowas wie:

executor.submit(() -> {
            while (running) {

                try {

                    Object obj = messages.take();
                   MailPushEventHandler m = (MailPushEventHandler) obj;
                   Object[] messages = m.getMessages();

 if (messages != null && messages.length > 0) {
            for (Object msg : messages) {
                try {
                    // Rufe andere EJB Klasse auf und speichere in DB
                } catch (MessagingException e) {
                    e.printStackTrace();
                } catch (IOException e) {

                    e.printStackTrace();
                }
            }
                } catch (InterruptedException e) {
                }
            }
        });
    }

 

[mihe7]

Die Klasse habe ich so angepasst:

Du musst die übergebene Queue natürlich noch in einer Instanzvariablen speichern.

public class MailPushEventHandler extends MailPushEmpfaenger {
    Queue<Object> messagesQueue;

    public MailPushEventHandler(Store store, Queue<Object> messagesQueue) {
        super(store);
        this.messageQueue = messagesQueue;
    }

    @Override
    public void onMailReceived(Object[] messages) {

        if (messages != null && messages.length > 0) {

            for (Object msg : messages) {
                messagesQueue.offer(msg);
            }
        }
    }
}

Mit all den "neuen" Email, die in Object[] messages sind, soll dann eben was passieren (Ticket in Datenbank anlegen).
Also eigentlich sowas wie:

Nein, das obj aus der Queue ist ja bereits eine neue Message (oder was auch immer Du in die Queue schiebst):

executor.submit(() -> {
            while (running) {
                try {
                    Object obj = messages.take();
                    // Rufe andere EJB Klasse auf und speichere obj in DB
                } catch (MessagingException e) {
                    e.printStackTrace();
                } catch (IOException e) {

                    e.printStackTrace();
                }
            }
                } catch (InterruptedException e) {
                }
            }
        });
    }

 

[beta20]

Ah ok, danke dir.

Dann bleibt jetzt aber noch die Frage wie ich die Überwachung von einzelnen Postfächer stoppe?
In der TicketEmailWatcherService schwebt mir sowas vor wie:

public void stop(String emailAddress){
???
}

Was ich brauche ist eine eindeutige ID / Emailadresse, in der ich das "watching" dann eines MailPushEventHandler stoppe?

 

[mihe7]

Was ich brauche ist eine eindeutige ID / Emailadresse, in der ich das "watching" dann eines MailPushEventHandler stoppe?

Du erzeugst ja für jede Adresse einen MailPushEventHandler, d. h. Du musst lediglich die Adresse über eine Map auf den Handler abbilden.

Also etwas wie:

private Map<String, MailPushEventHandler> handlers;

    @PostConstruct
    protected void initStartTicketListener() {

        LOGGER.info("START initStartTicketListener");

        handlers = new HashMap<>();
        messages = new LinkedBlockingQueue<>();
        running = true;

        try {

            List<TicketEmailSetting> list = ticketEmailSettingService.findAllTicketEmailSettingListByStatusActive();
            if (list == null || list.isEmpty())
                return;

            for (TicketEmailSetting ticketEmailSetting : list) {
                Store store = connect(ticketEmailSetting.getEmailSetting());
                handlers.put(ticketEmailSetting.getEmail(), new MailPushEventHandler(store, messages));
            }
        } catch (Exception e) {

        }
...

public void stop(String emailAddress){
    MailPushEventHandler handler = handlers.get(emailAddress);
    if (handler != null) {
        handler.stop();
        handlers.remove(emailAddress, handler);
    }
}

 

[beta20]

Danke, werde es mir später im Detail anschauen

 

[beta20]

Läuft nun super. Vielen vielen Dank.
Muss ich den Handler noch initialisieren? Wenn ja, wie?
Map<String, MailPushEventHandler> handlers


Ich habe noch ein Problem bei folgendem Code, welcher allerdings nur ein paar Mal auftritt.
Ich schätze, wenn ich die Email schon gelesen habe oder ähnliches.

        if (inbox.getMessageCount() > 0) {

            ArrayList<Message> msgs = new ArrayList<Message>();

            Message[] tempMsgs = inbox.getMessages(inbox.getMessageCount() - 10, inbox.getMessageCount());
            for (Message tempMsg : tempMsgs) {
                if (!tempMsg.isSet(Flags.Flag.SEEN)) {
                    msgs.add(tempMsg);
                    tempMsg.setFlag(Flags.Flag.SEEN, true);
                }
            }

Exception kommt hier: for (Message tempMsg : tempMsgs) {
Folgende Exception wird geworfen: javax.mail.MessageRemovedException

Wie könnte ich das noch beheben?

 

[mihe7]

Die Exception dürfte eine Zeile tiefer kommen. Du könntest vorher noch tempMsg.isExpunged() prüfen und eine ggf. doch auftretende Exception abfangen.

 

[beta20]

also du meinst:
if(tempMsg.isExpunged()) {
continue;
}

BTW:
Sehe gerade, dass hier dieser Code verwendet wird:

Message[] tempMsgs = inbox.getMessages(inbox.getMessageCount() - 10, inbox.getMessageCount());

Was macht das mit -10 ?
Das wurde doch nur aus Testzwecken so verwendet, dass eben nur die letzten 10 Email genommen werden?

 

[mihe7]

also du meinst:

Zum Beispiel oder

if (!tempMsg.isExpunged() && !tempMsg.isSet(Flags.Flag.SEEN)) {

Was macht das mit -10 ?
Das wurde doch nur aus Testzwecken so verwendet, dass eben nur die letzten 10 Email genommen werden?

Ja, die letzten 10 Emails. Warum: keine Ahnung, das ist nicht von mir

 

[beta20]

Hierzu eine generelle Frage:
- Wenn ich zB. 1000 Emailadressen gleichzeitig überwache, laufe ich hier gewiss in Performance - Probleme?

 

[mihe7]

Wenn ich zB. 1000 Emailadressen gleichzeitig überwache, laufe ich hier gewiss in Performance - Probleme?

Natürlich:

Außerdem ist es keine gute Idee, für jede Verbindung einen eigenen Thread zu verbraten.

 

[beta20]

Ok, danke....
D.h. man bräuchte genügend Ressourcen (Server) ?

 

[mihe7]

Nein, man braucht non-blocking I/O.