In this use case, the emails are a source system as any other. In a batch Dataflow, you read all emails and process them.
To do that, a new Datastore of type Adapter is added. Other than selecting the Jobserver and Adapter, there are no parameters to configure. The connection information is shared with the entire adapter instance.
Now you can import new "Documents". The way this adapter was implemented is that one Document is a list of emails matching your filter criterias, like the subject area containing a specific string.
This Document can now be dropped into any dataflow and processed further. It is higly nested, so make sure you are familiar with the NonRelationalDataModel in DI, meaning nesting, unnesting, schemas etc.