@Test public void testRun() throws Exception { final SystemService systemService = ctx().getBean("systemService", SystemService.class); final CustomerService customerService = ctx().getBean("customerService", CustomerService.class); final PromotionService promotionService = ctx().getBean("promotionService", PromotionService.class); final Runnable bulkCustomerTagProcessor = ctx().getBean("bulkCustomerTagProcessor", Runnable.class); final Customer customer1 = createCustomer(" cust1 "); final Customer customer2 = createCustomer(" cust2 "); final Customer customer3 = createCustomer(" cust3 "); final Promotion cust1Tag = promotionService.getGenericDao().getEntityFactory().getByIface(Promotion.class); cust1Tag.setCode("TAG1"); cust1Tag.setRank(10); cust1Tag.setShopCode("SHOIP1"); cust1Tag.setCurrency("EUR"); cust1Tag.setName("tag all cust1"); cust1Tag.setPromoType(Promotion.TYPE_CUSTOMER_TAG); cust1Tag.setPromoAction(Promotion.ACTION_TAG); cust1Tag.setEligibilityCondition("customer.firstname.indexOf(' cust1 ') != -1"); cust1Tag.setPromoActionContext("tag1"); cust1Tag.setCanBeCombined(true); cust1Tag.setEnabled(true); promotionService.create(cust1Tag); final Promotion cust3Tag = promotionService.getGenericDao().getEntityFactory().getByIface(Promotion.class); cust3Tag.setCode("TAG3"); cust3Tag.setRank(10); cust3Tag.setShopCode("SHOIP1"); cust3Tag.setCurrency("EUR"); cust3Tag.setName("tag all cust3"); cust3Tag.setPromoType(Promotion.TYPE_CUSTOMER_TAG); cust3Tag.setPromoAction(Promotion.ACTION_TAG); cust3Tag.setEligibilityCondition("customer.firstname.indexOf(' cust3 ') != -1"); cust3Tag.setPromoActionContext("tag3"); cust3Tag.setCanBeCombined(true); cust3Tag.setEnabled(true); promotionService.create(cust3Tag); final Promotion cust1or3Tag = promotionService.getGenericDao().getEntityFactory().getByIface(Promotion.class); cust1or3Tag.setCode("TAG1OR3"); cust1or3Tag.setRank(10); cust1or3Tag.setShopCode("SHOIP1"); cust1or3Tag.setCurrency("EUR"); cust1or3Tag.setName("tag all cust1 and cust3"); cust1or3Tag.setPromoType(Promotion.TYPE_CUSTOMER_TAG); cust1or3Tag.setPromoAction(Promotion.ACTION_TAG); cust1or3Tag.setEligibilityCondition("customer.firstname.indexOf(' cust1 ') != -1 || customer.firstname.indexOf(' cust3 ') != -1"); cust1or3Tag.setPromoActionContext("tag1or3"); cust1or3Tag.setCanBeCombined(true); cust1or3Tag.setEnabled(true); promotionService.create(cust1or3Tag); final Promotion cust3FirstTimeTag = promotionService.getGenericDao().getEntityFactory().getByIface(Promotion.class); cust3FirstTimeTag.setCode("TAG3FIRST"); cust3FirstTimeTag.setRank(10); cust3FirstTimeTag.setShopCode("SHOIP1"); cust3FirstTimeTag.setCurrency("EUR"); cust3FirstTimeTag.setName("tag all cust3 with no orders"); cust3FirstTimeTag.setPromoType(Promotion.TYPE_CUSTOMER_TAG); cust3FirstTimeTag.setPromoAction(Promotion.ACTION_TAG); cust3FirstTimeTag.setEligibilityCondition("customer.firstname.indexOf(' cust3 ') != -1 && customer.orders.isEmpty()"); cust3FirstTimeTag.setPromoActionContext("first3"); cust3FirstTimeTag.setCanBeCombined(true); cust3FirstTimeTag.setEnabled(true); promotionService.create(cust3FirstTimeTag); systemService.createOrGetAttributeValue("JOB_CUSTOMER_TAG_PAUSE", "Boolean"); systemService.updateAttributeValue("JOB_CUSTOMER_TAG_PAUSE", Boolean.FALSE.toString()); bulkCustomerTagProcessor.run(); systemService.updateAttributeValue("JOB_CUSTOMER_TAG_PAUSE", Boolean.TRUE.toString()); final Customer customer1tagged = customerService.findById(customer1.getCustomerId()); assertEquals("tag1 tag1or3", customer1tagged.getTag()); final Customer customer2tagged = customerService.findById(customer2.getCustomerId()); assertNull(customer2tagged.getTag()); final Customer customer3tagged = customerService.findById(customer3.getCustomerId()); assertEquals("first3 tag1or3 tag3", customer3tagged.getTag()); }
@Override public void run() { try { final int batchSize = determineBatchSize(); final List<Pair<Customer, String>> batch = new ArrayList<>(); customerService.findAllIterator(customer -> { final String tagsBefore = customer.getTag(); LOG.debug("Processing tagging for customer {} with tags {}", customer.getEmail(), tagsBefore); if (!customer.isGuest()) { for (final CustomerShop customerShop : customer.getShops()) { final Shop shop = shopService.getById(customerShop.getShop().getShopId()); for (final String currency : shop.getSupportedCurrenciesAsList()) { final PromotionContext promotionContext = promotionContextFactory.getInstance(shop.getCode(), currency); promotionContext.applyCustomerPromo(customer, null); } } if (!StringUtils.equals(tagsBefore, customer.getTag())) { if (batch.size() + 1 >= batchSize) { self().updateCustomers(batch); listener.count(UPDATED_COUNTER, batch.size()); batch.clear(); customerService.getGenericDao().clear(); } batch.add(new Pair<>(customer, customer.getTag())); } listener.count(PROCESS_COUNTER); listener.notifyPing("Processed {}, updated: {}", listener.getCount(PROCESS_COUNTER), listener.getCount(UPDATED_COUNTER)); } return true; }); if (batch.size() > 0) { self().updateCustomers(batch); listener.count(UPDATED_COUNTER, batch.size()); } } catch (Exception exp){ LOG.error(Markers.alert(), "Processing tagging for customer error: " + exp.getMessage(), exp); throw new RuntimeException(exp); } listener.notifyCompleted(); listener.reset(); }
BulkCustomerTagProcessorImpl implements BulkCustomerTagProcessorInternal, JobStatusAware { @Override public void run() { try { final int batchSize = determineBatchSize(); final List<Pair<Customer, String>> batch = new ArrayList<>(); customerService.findAllIterator(customer -> { final String tagsBefore = customer.getTag(); LOG.debug("Processing tagging for customer {} with tags {}", customer.getEmail(), tagsBefore); if (!customer.isGuest()) { for (final CustomerShop customerShop : customer.getShops()) { final Shop shop = shopService.getById(customerShop.getShop().getShopId()); for (final String currency : shop.getSupportedCurrenciesAsList()) { final PromotionContext promotionContext = promotionContextFactory.getInstance(shop.getCode(), currency); promotionContext.applyCustomerPromo(customer, null); } } if (!StringUtils.equals(tagsBefore, customer.getTag())) { if (batch.size() + 1 >= batchSize) { self().updateCustomers(batch); listener.count(UPDATED_COUNTER, batch.size()); batch.clear(); customerService.getGenericDao().clear(); } batch.add(new Pair<>(customer, customer.getTag())); } listener.count(PROCESS_COUNTER); listener.notifyPing("Processed {}, updated: {}", listener.getCount(PROCESS_COUNTER), listener.getCount(UPDATED_COUNTER)); } return true; }); if (batch.size() > 0) { self().updateCustomers(batch); listener.count(UPDATED_COUNTER, batch.size()); } } catch (Exception exp){ LOG.error(Markers.alert(), "Processing tagging for customer error: " + exp.getMessage(), exp); throw new RuntimeException(exp); } listener.notifyCompleted(); listener.reset(); } }
BulkCustomerTagProcessorImpl implements BulkCustomerTagProcessorInternal, JobStatusAware { @Override public void run() { try { final int batchSize = determineBatchSize(); final List<Pair<Customer, String>> batch = new ArrayList<>(); customerService.findAllIterator(customer -> { final String tagsBefore = customer.getTag(); LOG.debug("Processing tagging for customer {} with tags {}", customer.getEmail(), tagsBefore); if (!customer.isGuest()) { for (final CustomerShop customerShop : customer.getShops()) { final Shop shop = shopService.getById(customerShop.getShop().getShopId()); for (final String currency : shop.getSupportedCurrenciesAsList()) { final PromotionContext promotionContext = promotionContextFactory.getInstance(shop.getCode(), currency); promotionContext.applyCustomerPromo(customer, null); } } if (!StringUtils.equals(tagsBefore, customer.getTag())) { if (batch.size() + 1 >= batchSize) { self().updateCustomers(batch); listener.count(UPDATED_COUNTER, batch.size()); batch.clear(); customerService.getGenericDao().clear(); } batch.add(new Pair<>(customer, customer.getTag())); } listener.count(PROCESS_COUNTER); listener.notifyPing("Processed {}, updated: {}", listener.getCount(PROCESS_COUNTER), listener.getCount(UPDATED_COUNTER)); } return true; }); if (batch.size() > 0) { self().updateCustomers(batch); listener.count(UPDATED_COUNTER, batch.size()); } } catch (Exception exp){ LOG.error(Markers.alert(), "Processing tagging for customer error: " + exp.getMessage(), exp); throw new RuntimeException(exp); } listener.notifyCompleted(); listener.reset(); } BulkCustomerTagProcessorImpl(final ShopService shopService,
                                        final CustomerService customerService,
                                        final SystemService systemService,
                                        final PromotionContextFactory promotionContextFactory); }
BulkCustomerTagProcessorImpl implements BulkCustomerTagProcessorInternal, JobStatusAware { @Override public void run() { try { final int batchSize = determineBatchSize(); final List<Pair<Customer, String>> batch = new ArrayList<>(); customerService.findAllIterator(customer -> { final String tagsBefore = customer.getTag(); LOG.debug("Processing tagging for customer {} with tags {}", customer.getEmail(), tagsBefore); if (!customer.isGuest()) { for (final CustomerShop customerShop : customer.getShops()) { final Shop shop = shopService.getById(customerShop.getShop().getShopId()); for (final String currency : shop.getSupportedCurrenciesAsList()) { final PromotionContext promotionContext = promotionContextFactory.getInstance(shop.getCode(), currency); promotionContext.applyCustomerPromo(customer, null); } } if (!StringUtils.equals(tagsBefore, customer.getTag())) { if (batch.size() + 1 >= batchSize) { self().updateCustomers(batch); listener.count(UPDATED_COUNTER, batch.size()); batch.clear(); customerService.getGenericDao().clear(); } batch.add(new Pair<>(customer, customer.getTag())); } listener.count(PROCESS_COUNTER); listener.notifyPing("Processed {}, updated: {}", listener.getCount(PROCESS_COUNTER), listener.getCount(UPDATED_COUNTER)); } return true; }); if (batch.size() > 0) { self().updateCustomers(batch); listener.count(UPDATED_COUNTER, batch.size()); } } catch (Exception exp){ LOG.error(Markers.alert(), "Processing tagging for customer error: " + exp.getMessage(), exp); throw new RuntimeException(exp); } listener.notifyCompleted(); listener.reset(); } BulkCustomerTagProcessorImpl(final ShopService shopService,
                                        final CustomerService customerService,
                                        final SystemService systemService,
                                        final PromotionContextFactory promotionContextFactory); @Override JobStatus getStatus(final String token); @Override void run(); @Override void updateCustomers(final List<Pair<Customer, String>> customers); void setBatchSize(final int batchSize); BulkCustomerTagProcessorInternal getSelf(); }
BulkCustomerTagProcessorImpl implements BulkCustomerTagProcessorInternal, JobStatusAware { @Override public void run() { try { final int batchSize = determineBatchSize(); final List<Pair<Customer, String>> batch = new ArrayList<>(); customerService.findAllIterator(customer -> { final String tagsBefore = customer.getTag(); LOG.debug("Processing tagging for customer {} with tags {}", customer.getEmail(), tagsBefore); if (!customer.isGuest()) { for (final CustomerShop customerShop : customer.getShops()) { final Shop shop = shopService.getById(customerShop.getShop().getShopId()); for (final String currency : shop.getSupportedCurrenciesAsList()) { final PromotionContext promotionContext = promotionContextFactory.getInstance(shop.getCode(), currency); promotionContext.applyCustomerPromo(customer, null); } } if (!StringUtils.equals(tagsBefore, customer.getTag())) { if (batch.size() + 1 >= batchSize) { self().updateCustomers(batch); listener.count(UPDATED_COUNTER, batch.size()); batch.clear(); customerService.getGenericDao().clear(); } batch.add(new Pair<>(customer, customer.getTag())); } listener.count(PROCESS_COUNTER); listener.notifyPing("Processed {}, updated: {}", listener.getCount(PROCESS_COUNTER), listener.getCount(UPDATED_COUNTER)); } return true; }); if (batch.size() > 0) { self().updateCustomers(batch); listener.count(UPDATED_COUNTER, batch.size()); } } catch (Exception exp){ LOG.error(Markers.alert(), "Processing tagging for customer error: " + exp.getMessage(), exp); throw new RuntimeException(exp); } listener.notifyCompleted(); listener.reset(); } BulkCustomerTagProcessorImpl(final ShopService shopService,
                                        final CustomerService customerService,
                                        final SystemService systemService,
                                        final PromotionContextFactory promotionContextFactory); @Override JobStatus getStatus(final String token); @Override void run(); @Override void updateCustomers(final List<Pair<Customer, String>> customers); void setBatchSize(final int batchSize); BulkCustomerTagProcessorInternal getSelf();  }