<commit_before>package nl.cerios.demo.processor;
import io.reactivex.Flowable;
import io.reactivex.processors.ReplayProcessor;
import nl.cerios.demo.http.HttpRequestData;
import nl.cerios.demo.http.PurchaseHttpHandler;
import nl.cerios.demo.service.CustomerData;
import nl.cerios.demo.service.LocationConfig;
import nl.cerios.demo.service.PurchaseRequest;
import nl.cerios.demo.service.ValidationException;


public class PurchaseRequestProcessor_Rx extends BaseProcessor {

	public Flowable<CustomerData> handle(HttpRequestData requestData, PurchaseHttpHandler purchaseHandler)
	{

		Flowable<PurchaseRequest> purchaseRequestObs = purchaseRequestController.getPurchaseRequest_Rx( requestData.getPurchaseRequestId());
		purchaseRequestObs.onErrorResumeNext( (e) -> {
			if (e instanceof ValidationException) {
				purchaseHandler.notifyValidationError( e.getMessage());
			}
			return Flowable.error( e);
		});
		ReplayProcessor<PurchaseRequest> purchaseRequestPublisher= ReplayProcessor.create();
		purchaseRequestObs.subscribe( purchaseRequestPublisher);


		Flowable<CustomerData> customerData= customerService.getCustomerData_Rx( purchaseRequestPublisher
				.map( purchaseRequest-> purchaseRequest.getCustomerId()));
		/*} catch (ValidationException e) {
			purchaseHandler.notifyValidationError( e.getMessage());
			return;
		}*/
		Flowable<Integer> locationIdObs= purchaseRequestPublisher
				.map( purchaseRequest1-> purchaseRequest1.getLocationId());

		//Flowable<LocationConfig> locationDataObs= locationService_Rx.getLocationConfig(locationIdObs.blockingFirst());
		return customerData;

		/*
		// Now there is a customer, we can store it in the speed layer
		purchaseRequestController.store( purchaseRequest);

		CustomerValidation customerValidation = customerService.validateCustomer( customerData, locationData);
		if (customerValidation.getStatus() != Status.OK) {
			purchaseHandler.notifyValidationError( customerValidation.getMessage());
			return;
		}

		TransactionValidation transactionValidation = transactionService.validate_Sync( purchaseRequest, customerData);
		if (transactionValidation.getStatus() != Status.OK) {
			purchaseHandler.notifyValidationError( transactionValidation.getMessage());
		}

		OrderData orderData= orderService.createOrder( purchaseRequest);
		purchaseRequestController.update( purchaseRequest, orderData);

		Status status= transactionService.linkOrderToTransaction_Sync( purchaseRequest);
		if (status != Status.OK) {

			// Payment and order OK, however: automatic linking failed --> manual 
			mailboxHandler.sendMessage( composeLinkingFailedMessage( purchaseRequest));
		}

		purchaseHandler.notifyComplete( purchaseRequest);
		 */
	}

}
<commit_msg>flowable applied<commit_after>package nl.cerios.demo.processor;
import io.reactivex.Flowable;
import io.reactivex.processors.ReplayProcessor;
import nl.cerios.demo.http.HttpRequestData;
import nl.cerios.demo.http.PurchaseHttpHandler;
import nl.cerios.demo.service.CustomerData;
import nl.cerios.demo.service.CustomerValidation;
import nl.cerios.demo.service.LocationConfig;
import nl.cerios.demo.service.PurchaseRequest;
import nl.cerios.demo.service.Status;
import nl.cerios.demo.service.ValidationException;


public class PurchaseRequestProcessor_Rx extends BaseProcessor {

	public Flowable<CustomerData> handle(HttpRequestData requestData, PurchaseHttpHandler purchaseHandler)
	{

		Flowable<PurchaseRequest> purchaseRequestObs = 
				purchaseRequestController.getPurchaseRequest_Rx( requestData.getPurchaseRequestId())
				.onErrorResumeNext( (e) -> {
					if (e instanceof ValidationException) {
						purchaseHandler.notifyValidationError( e.getMessage());
					}
					return Flowable.error( e);
				});
		ReplayProcessor<PurchaseRequest> purchaseRequestPublisher= ReplayProcessor.create();
		purchaseRequestObs.subscribe( purchaseRequestPublisher);


		Flowable<CustomerData> customerDataObs= customerService.getCustomerData_Rx( purchaseRequestPublisher
				.map( purchaseRequest-> purchaseRequest.getCustomerId()));
		customerDataObs.onErrorResumeNext( (e) -> {
			if (e instanceof ValidationException) {
				purchaseHandler.notifyValidationError( e.getMessage());
			}
			return Flowable.error( e);
		});

		Flowable<LocationConfig> locationDataObs= purchaseRequestPublisher
				.map( purchaseRequest1-> purchaseRequest1.getLocationId())
				.flatMap( locationId -> locationService_Rx.getLocationConfig(locationId));

		Flowable<CustomerValidation> customerValidationObs= Flowable.zip( customerDataObs, locationDataObs,
				(customerData, locationData) -> customerService.validateCustomer( customerData, locationData))
				.map( customerValidation -> {
					if (customerValidation.getStatus() != Status.OK) {
						purchaseHandler.notifyValidationError( customerValidation.getMessage());
						throw new ValidationException("Validation Error");
					}
					return customerValidation;
				});

		return Flowable.merge( customerDataObs, customerValidationObs.flatMap( l->Flowable.empty()));
		/*
		// Now there is a customer, we can store it in the speed layer
		purchaseRequestController.store( purchaseRequest);
		TransactionValidation transactionValidation = transactionService.validate_Sync( purchaseRequest, customerData);
		if (transactionValidation.getStatus() != Status.OK) {
			purchaseHandler.notifyValidationError( transactionValidation.getMessage());
		}

		OrderData orderData= orderService.createOrder( purchaseRequest);
		purchaseRequestController.update( purchaseRequest, orderData);

		Status status= transactionService.linkOrderToTransaction_Sync( purchaseRequest);
		if (status != Status.OK) {

			// Payment and order OK, however: automatic linking failed --> manual 
			mailboxHandler.sendMessage( composeLinkingFailedMessage( purchaseRequest));
		}

		purchaseHandler.notifyComplete( purchaseRequest);
		 */
	}

}

{'fraction_non_alphanumeric': 0.09482316760635572, 'fraction_numerical': 0.0006834102169827439, 'mean_word_length': 6.070048309178744, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 28, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7061989', 'n_tokens_mistral': 1737, 'n_tokens_neox': 1527, 'n_words': 299}