Problem with entity manager in Spring Webflux

In my web client, where I started using Spring WebFlux, I got following exception:

reactor.core.Exceptions$ErrorCallbackNotImplemented: javax.persistence.TransactionRequiredException: No EntityManager with actual transaction available for current thread - cannot reliably process 'persist' call
Caused by: javax.persistence.TransactionRequiredException: No EntityManager with actual transaction available for current thread - cannot reliably process 'persist' call

Code:

@Service
@Transactional
@Slf4j
public class PersonSyncServiceImpl
		extends HotelApiCommunicationService implements PersonSyncService {

	private static final String REST_ENDPOINT_PATH = "/api/sync/person";

	@PersistenceContext
	private EntityManager em;
	@Autowired
	private SynchronizationService syncMgr;


	@Override
	public int initialSyncPersonData(ObiektDTO obiekt) {

		WebClient client = WebClient.create("http://" + obiekt.getAdresIp());
		AtomicInteger count = new AtomicInteger(0);

		Flux<Person> personFlux = client.get()
				.uri(REST_ENDPOINT_PATH)
				.retrieve()
				.bodyToFlux(Person.class);

		personFlux.subscribe(person -> {
			CNPerson cnPerson = syncMgr.convertObject(person, CNPerson.class);
			cnPerson.setObiektLoid(obiekt.getLoid());
			em.persist(cnPerson);
			count.getAndIncrement();
		});

		return count.get();
	}
}

I know that the problem is on this line, because reactor can't get entity manager.

em.persist(cnPerson);

Controller method on server:

	@GetMapping(value = "/person", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
	public Flux<Person> getPersonList() {

		return Flux.fromStream(
				personDao.findAll().stream());
	}

How to fix it? How to access transaction with reactor and using JPA save a record to database?

First of all, Spring Data R2DBC doesn't work with JPA/Hibernate and blocking JDBC Driver. So either, you need to get rid of webflux or JPA.

1. With JPA

If suppose, you decide to proceed with JPA, then you can call the Person Sync API using Spring RestTemplate or WebClient Blocking call like this:

   List<Person> persons = client.get()
                .uri(REST_ENDPOINT_PATH)
                .retrieve()
                .bodyToFlux(Person.class)
                .collectList()
                .block();

Transform them like you are doing;

List<CNPerson> cnPersons = new ArrayList();
persons.forEach(person -> {
  CNPerson cnPerson = syncMgr.convertObject(person, CNPerson.class);
  cnPerson.setObiektLoid(obiekt.getLoid());
  cnPersons.add(cnPerson);
});

And then save them all in one shot, using your org.springframework.data.jpa.repository.JpaRepository.saveAll() [For this, you need to create a Repository for the CNPerson Entity.]

Or manually manage the transaction and save each CNPerson within it.

Lastly, to answer your question Is there other method to get a stream on client side? - Answer is NO if you want to do it in one transaction. Coz you have to load the entire response of the Person Sync API in memory.

2. Without JPA (Using Spring Webflux and R2DBC)

You can still have transactional with Spring Data R2DBC. You can read more about it here. Note that with R2DBC, you have to use the Non-blocking drivers for whatever type of database you are using. A list is given here.

If you use JPA in a reactive context and you don't need to execute all your persistance operations inside a transaction, you can try to manually create a new transaction for each operation.

1 - Inject a PlatformTransactionManager

@Autowired
private PlatformTransactionManager transactionManager;

2 - Create a new TransactionTemplate (this could be done in constructor)

this.transactionTemplate = new TransactionTemplate(transactionManager);

3 - Execute your persistent calls inside a call to transactionTemplate.execute :

personFlux.subscribe(person -> 
  transactionTemplate.execute(transactionStatus -> {
            CNPerson cnPerson = syncMgr.convertObject(person, CNPerson.class);
            cnPerson.setObiektLoid(obiekt.getLoid());
            em.persist(cnPerson);
            count.getAndIncrement();
});

Note that this will create a new transaction for each call to em.persist.