Apache Camel을 이용한 대출 모집인 구현

.eip-chapter-9

Apache Camel을 이용한 대출 모집인 구현 인용

BankQouteService.java

package com.brm.pm.bank;

import com.brm.pm.data.BankReply;
import com.brm.pm.data.BankRequest;

public interface BankQouteService {

	public BankReply computeBankReply(BankRequest bankRequest);

}

BankQouteServiceImp.java

package com.brm.pm.bank;

import com.brm.pm.data.BankReply;
import com.brm.pm.data.BankRequest;

import java.util.concurrent.atomic.AtomicInteger;

public class BankQouteServiceImp implements BankQouteService {

	final static Integer MAX_LOAN_TERM = 360;

	String bankName;
	private static AtomicInteger qouteCount = new AtomicInteger();
	Double primeRate;
	Double ratePremium;

	@Override
	public BankReply computeBankReply(BankRequest bankRequest) {

		BankReply bankReply = new BankReply();

		if (bankRequest.LoanTerm <= MAX_LOAN_TERM) {
			bankReply.InterestRate = primeRate + ratePremium
					+ (bankRequest.LoanTerm / 12.0) / 10
					+ Math.random();
			bankReply.ErrorCode = 0;
		} else {
			bankReply.InterestRate = 0.0;
			bankReply.ErrorCode = 1;
		}
		bankReply.QuoteId = String.format("%s : %06d", bankName, qouteCount.addAndGet(1));
		return bankReply;
	}

	public void setBankName(String bankName) {
		this.bankName = bankName;
	}

	public void setPrimeRate(Double primeRate) {
		this.primeRate = primeRate;
	}

	public void setRatePremium(Double ratePremium) {
		this.ratePremium = ratePremium;
	}

}

BankReplyAggregator.java

package com.brm.pm.process;

import static com.brm.pm.data.Constants.BANK_REQUEST_CORRELATION_ID;
import com.brm.pm.data.BankReply;
import com.brm.pm.data.Job;
import com.brm.pm.job.JobRepository;
import org.apache.camel.Exchange;
import org.apache.camel.processor.aggregate.AggregationStrategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

public class BankReplyAggregator implements AggregationStrategy {

	private static final Logger logger = LoggerFactory.getLogger(BankReplyAggregator.class);

	@Autowired
	JobRepository jobRepository;

	@Override
	public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {

		String correlationId = newExchange.getIn().getHeader(BANK_REQUEST_CORRELATION_ID, String.class);
		BankReply bankReply = newExchange.getIn().getBody(BankReply.class);

		try {
			Job job = jobRepository.getJob(correlationId);
			job.bids.add(bankReply);

		} catch (Exception e) {
			logger.warn(String.format("Could not find Job Object with ID (%s)", correlationId), e);
		}

		if (oldExchange == null) {
			return newExchange;
		}
		return oldExchange;
	}

}

banks.xml

...
		<route id="Bank1Route">
			<from uri="jms:topic:bank.loan.request" />
			<to uri="bean:bank1" />
			<to uri="jms:queue:bank.loan.reply" />
		</route>
		<route id="Bank2Route">
			<from uri="jms:topic:bank.loan.request" />
			<to uri="bean:bank2" />
			<to uri="jms:queue:bank.loan.reply" />
		</route>
		<route id="Bank3Route">
			<from uri="jms:topic:bank.loan.request" />
			<to uri="bean:bank3" />
			<to uri="jms:queue:bank.loan.reply" />
		</route>
		<route id="Bank4Route">
			<from uri="jms:topic:bank.loan.request" />
			<to uri="bean:bank4" />
			<to uri="jms:queue:bank.loan.reply" />
		</route>
		<route id="Bank5Route">
			<from uri="jms:topic:bank.loan.request" />
			<to uri="bean:bank5" />
			<to uri="jms:queue:bank.loan.reply" />
		</route>
...
	<bean id="bank1" class="com.brm.pm.bank.BankQouteServiceImp" p:bankName="Bank1" p:primeRate="0.1" p:ratePremium="0.1" />
	<bean id="bank2" class="com.brm.pm.bank.BankQouteServiceImp" p:bankName="Bank2" p:primeRate="0.1" p:ratePremium="0.1" />
	<bean id="bank3" class="com.brm.pm.bank.BankQouteServiceImp" p:bankName="Bank3" p:primeRate="0.1" p:ratePremium="0.1" />
	<bean id="bank4" class="com.brm.pm.bank.BankQouteServiceImp" p:bankName="Bank4" p:primeRate="0.1" p:ratePremium="0.1" />
	<bean id="bank5" class="com.brm.pm.bank.BankQouteServiceImp" p:bankName="Bank5" p:primeRate="0.1" p:ratePremium="0.1" />
...	

borkerProcessManager.full.xml

...
	<cxf:cxfEndpoint id="brokerQouteServiceEndpoint" address="http://localhost:1234/broker/loanQouteService"
		serviceClass="com.brm.pm.broker.BrokerQouteService">
	</cxf:cxfEndpoint>
	<bean id="brokerQouteServiceImp" class="com.brm.pm.broker.BrokerQouteServiceImp" />

	<cxf:cxfEndpoint id="creditScoreServiceEndpoint" address="http://localhost:1234/creditBuruea/CreditScoreService"
		serviceClass="com.brm.pm.creditburea.CreditScoreService">
	</cxf:cxfEndpoint>
	<bean id="creditScoreServiceImp" class="com.brm.pm.creditburea.CreditScoreServiceImp" />
...
	<bean id="brokerQouteServiceImp" class="com.brm.pm.broker.BrokerQouteServiceImp" />
...
	<route id="borkerProcessManager">
		<from uri="cxf:bean:brokerQouteServiceEndpoint" />
		<to uri="bean:brokerQouteService" />
	</route>
...

borkerProcessManager.xml

		<!-- 프로세스 관리자 라우팅 -->
		<route id="borkerProcessManager">
			<from uri="cxf:bean:brokerQouteServiceEndpoint" />
			<to uri="bean:brokerQouteServiceImp" />
		</route>

BrokerQouteService.java

package com.brm.pm.broker;

import com.brm.pm.data.LoanQouteReply;
import com.brm.pm.data.LoanQouteRequest;

public interface BrokerQouteService {

	public LoanQouteReply service(LoanQouteRequest loanQouteRequest) throws Exception;

}

brokerQouteServiceEndpoint.xml

  	<!-- 대출 모집인 웹 서비스 -->
	<cxf:cxfEndpoint id="brokerQouteServiceEndpoint" address="http://localhost:1234/broker/loanQouteService"
		serviceClass="com.brm.pm.broker.BrokerQouteService">
	</cxf:cxfEndpoint>
	<bean id="brokerQouteServiceImp" class="com.brm.pm.broker.BrokerQouteServiceImp" />

BrokerQouteServiceImp.java

package com.brm.pm.broker;

import static com.brm.pm.data.Constants.BANK_REQUEST_CORRELATION_ID;
import static java.util.UUID.randomUUID;
import com.brm.pm.data.BankReply;
import com.brm.pm.data.BankRequest;
import com.brm.pm.data.CreditBureaReply;
import com.brm.pm.data.CreditBureaRequest;
import com.brm.pm.data.Job;
import com.brm.pm.data.LoanQouteReply;
import com.brm.pm.data.LoanQouteRequest;
import com.brm.pm.job.JobRepository;

import java.util.ArrayList;
import java.util.Collections;

import org.apache.camel.Produce;
import org.apache.camel.ProducerTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

public class BrokerQouteServiceImp implements BrokerQouteService {
	/**
	 * Logger for this class
	 */
	private static final Logger logger = LoggerFactory.getLogger(BrokerQouteServiceImp.class);

	//  Credit Burea Web Interface
	@Produce(uri = "cxf:bean:creditScoreServiceEndpoint?defaultOperationName=getCreditScore")
	ProducerTemplate creditBurea;

	// Bank Publish Subscribe Channel
	@Produce(uri = "jms:topic:bank.loan.request")
	ProducerTemplate auction;

	// Job Repository
	@Autowired
	JobRepository jobRepository;

	@Override
	public LoanQouteReply service(LoanQouteRequest loanQouteRequest) throws Exception {

		logger.info("service called");

		// Create Job
		Job job = new Job(randomUUID().toString());
		job.request = loanQouteRequest;

		// Create AE Object Credit Request
		CreditBureaRequest credit = new CreditBureaRequest();
		credit.SSN = job.request.SSN;
		job.creditRequest = credit;

		// Invoke Synchronous Credit Burea Request
		job.creditReply = creditBurea.requestBody(job.creditRequest, CreditBureaReply.class);

		// Create AE Object Bank Request
		BankRequest bank = new BankRequest();
		bank.CorrelationtionID = job.getId();
		bank.SSN = job.request.SSN;
		bank.CreditScore = job.creditReply.CreditScore;
		bank.HistoryLength = job.creditReply.HistoryLength;
		bank.LoanAmount = job.request.LoanAmount;
		bank.LoanTerm = job.request.LoanTerm;
		job.bankRequest = bank;

		// Inintialize Bid Array
		job.bids = new ArrayList<BankReply>();
		job.bids = Collections.synchronizedList(new ArrayList<BankReply>());

		// Save Job to Repository
		jobRepository.putJob(job.bankRequest.CorrelationtionID, job);

		// Publish Asynchronous Bank Request Message
		auction.sendBodyAndHeader(job.bankRequest,
				BANK_REQUEST_CORRELATION_ID, job.bankRequest.CorrelationtionID);

		// Wait For Bank Relpy And Filter On CorrelationID
		synchronized (job) {
			job.wait();
		}

		return job.reply;
	}
}

brokerReplyManager.full.xml

...
	<bean id="bankReplyAggregator" class="com.brm.pm.process.BankReplyAggregator" />
	<bean id="processReceivedQoute" class="com.brm.pm.process.ProcessReceivedQoute" />
...
	<route id="brokerReplyManager">
		<from uri="jms:queue:bank.loan.reply" />
		<aggregate completionTimeout="10000" strategyRef="bankReplyAggregator">
			<correlationExpression>
				<simple>header.BankRequestCorrelatonID</simple>
			</correlationExpression>
			<to uri="bean:ProcessReceivedQoute" />
		</aggregate>
	</route>
...

brokerReplyManager.xml

		<!-- 응답 관리자 라우팅 -->
		<route id="brokerReplyManager">
			<from uri="jms:queue:bank.loan.reply" />
			<aggregate completionTimeout="10000" strategyRef="bankReplyAggregator">
				<correlationExpression>
					<simple>header.BankRequestCorrelatonID</simple>
				</correlationExpression>
				<to uri="bean:processReceivedQoute" />
			</aggregate>
		</route>

client.xml

...
  <bean id="client" class="com.brm.pm.client.LoanQouteClient" />
...
  <route>
	  <from uri="timer:client?period=5000" />
	  <to uri="bean:client" />
  </route>
...

CreditScoreService.java

package com.brm.pm.creditburea;

import com.brm.pm.data.CreditBureaReply;
import com.brm.pm.data.CreditBureaRequest;

public interface CreditScoreService {

	public CreditBureaReply getCreditScore(CreditBureaRequest creditRequest);

}

creditScoreServiceEndpoint.xml

	<!-- 신용 평가 기관 웹 서비스 -->
	<cxf:cxfEndpoint id="creditScoreServiceEndpoint" address="http://localhost:1234/creditBuruea/CreditScoreService"
		serviceClass="com.brm.pm.creditburea.CreditScoreService">
	</cxf:cxfEndpoint>
	<bean id="creditScoreServiceImp" class="com.brm.pm.creditburea.CreditScoreServiceImp" />

CreditScoreServiceImp.java

package com.brm.pm.creditburea;

import com.brm.pm.data.CreditBureaReply;
import com.brm.pm.data.CreditBureaRequest;

public class CreditScoreServiceImp implements CreditScoreService {

	@Override
	public CreditBureaReply getCreditScore(CreditBureaRequest creditRequest) {
		CreditBureaReply creditReply = new CreditBureaReply();

		creditReply.SSN = creditRequest.SSN;
		creditReply.CreditScore = (int) (Math.random() * 600 + 300);
		creditReply.HistoryLength = (int) (Math.random() * 19 + 1);

		return creditReply;
	}

}

EhCacheRepository.java

package com.brm.pm.job;

import static org.apache.camel.component.cache.CacheConstants.CACHE_KEY;
import static org.apache.camel.component.cache.CacheConstants.CACHE_OPERATION;
import static org.apache.camel.component.cache.CacheConstants.CACHE_OPERATION_ADD;
import static org.apache.camel.component.cache.CacheConstants.CACHE_OPERATION_GET;
import com.brm.pm.data.Job;

import java.util.HashMap;
import java.util.Map;

import org.apache.camel.Produce;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.component.cache.CacheConstants;

public class EhCacheRepository implements JobRepository {

	@Produce
	ProducerTemplate jobSlot;

	@Override
	public Job getJob(String objectId) throws Exception {
		Map<String, Object> headers = new HashMap<String, Object>();
		headers.put(CacheConstants.CACHE_OPERATION, CACHE_OPERATION_GET);
		headers.put(CACHE_KEY, objectId);
		return jobSlot.requestBodyAndHeaders("direct:jobslot", this, headers, Job.class);
	}

	@Override
	public void putJob(String objectId, Job job) throws Exception {
		Map<String, Object> headers = new HashMap<String, Object>();
		headers.put(CACHE_OPERATION, CACHE_OPERATION_ADD);
		headers.put(CACHE_KEY, objectId);
		jobSlot.requestBodyAndHeaders("direct:jobslot", job, headers);
	}
}

JobRepository.java

package com.brm.pm.job;

import com.brm.pm.data.Job;


public interface JobRepository {

	public Job getJob(String key) throws Exception;

	public void putJob(String key, Job job) throws Exception;
}

JobSloteRoute.xml

...
	<bean id="jobRepository" class="com.brm.pm.job.EhCacheRepository" />
...
	<route id="JobSloteRoute">
		<from uri="direct:jobslot" />
		<to uri="cache://JobSlot" />
	</route>
...

LoanQouteClient.java

package com.brm.pm.client;

import com.brm.pm.data.LoanQouteReply;
import com.brm.pm.data.LoanQouteRequest;
import org.apache.camel.Produce;
import org.apache.camel.ProducerTemplate;
import org.apache.commons.lang3.builder.ReflectionToStringBuilder;

public class LoanQouteClient {

	@Produce(uri = "cxf:bean:brokerQouteServiceEndpoint?defaultOperationName=service")
	ProducerTemplate client;

	public void request() {

		// 요청 파라미터 입력
		LoanQouteRequest loanQoute = new LoanQouteRequest();
		loanQoute.SSN = "1234567890";
		loanQoute.LoanAmount = 100000.0;
		loanQoute.LoanTerm = 360;

		// 대출 견적 요청
		LoanQouteReply result = client.requestBody(loanQoute, LoanQouteReply.class);

		// 결과 표시
		System.err.println("result = " + ReflectionToStringBuilder.toString(result));
	}
}

ProcessReceivedQoute.java

package com.brm.pm.process;

import com.brm.pm.data.BankReply;
import com.brm.pm.data.Job;
import com.brm.pm.data.LoanQouteReply;
import com.brm.pm.job.JobRepository;
import org.apache.camel.Header;
import org.springframework.beans.factory.annotation.Autowired;

public class ProcessReceivedQoute {

	@Autowired
	JobRepository jobRepository;

	public void process(@Header("BankRequestCorrelatonID") String correlationId) throws Exception {

		try {
			Job job = jobRepository.getJob(correlationId);

			LoanQouteReply loanReply = new LoanQouteReply();

			loanReply.SSN = job.request.SSN;
			loanReply.LoanAmount = job.request.LoanAmount;

			for (BankReply bid : job.bids) {
				if (bid.InterestRate < loanReply.InterestRate) {
					loanReply.InterestRate = bid.InterestRate;
					loanReply.QouteID = bid.QuoteId;
				}
			}
			job.reply = loanReply;
			synchronized (job) {
				job.notify();
			}
		} catch (Exception e) {
			throw new Exception(String.format("Could not find Job Object with ID (%s)", correlationId));
		}
	}
}

ProducerTemplate.auction.java

package com.brm.pm.broker;
...
public class BrokerQouteServiceImp implements BrokerQouteService {
...
	// Bank Publish Subscribe Channel
	@Produce(uri = "jms:topic:bank.loan.request")
	ProducerTemplate auction;
...
	@Override
	public LoanQouteReply service(LoanQouteRequest loanQouteRequest) throws Exception {
...
		// Publish Asynchronous Bank Request Message
		auction.sendBodyAndHeader(job.bankRequest,
				BANK_REQUEST_CORRELATION_ID, job.bankRequest.CorrelationtionID);
...
	}
}

ProducerTemplate.creditBurea.java

package com.brm.pm.broker;
...
public class BrokerQouteServiceImp implements BrokerQouteService {

	//  Credit Burea Web Interface
	@Produce(uri = "cxf:bean:creditScoreServiceEndpoint?defaultOperationName=getCreditScore")
	ProducerTemplate creditBurea;
...
	@Override
	public LoanQouteReply service(LoanQouteRequest loanQouteRequest) throws Exception {
...
		// Invoke Synchronous Credit Burea Request
		job.creditReply = creditBurea.requestBody(job.creditRequest, CreditBureaReply.class);
...
	}
}

result

[main] MainSupport                    INFO  Apache Camel 2.13.2 starting
[main] BrokerService                  INFO  Using Persistence Adapter: MemoryPersistenceAdapter
[main] BrokerService                  INFO  Apache ActiveMQ 5.9.0 (eip-chapter-9, ID:JHCHA-PC-30971-1416318377535-0:1) is starting
[main] TransportServerThreadSupport   INFO  Listening for connections at: tcp://127.0.0.1:61616
[main] TransportConnector             INFO  Connector tcp://localhost:61616 started
[main] BrokerService                  INFO  Apache ActiveMQ 5.9.0 (eip-chapter-9, ID:JHCHA-PC-30971-1416318377535-0:1) started
[main] BrokerService                  INFO  For help or more information please see: http://activemq.apache.org
[main] DefaultTypeConverter           INFO  Loaded 197 type converters
[main] SpringCamelContext             INFO  Apache Camel 2.13.2 (CamelContext: camel-1) is starting
[main] ManagedManagementStrategy      INFO  JMX is enabled
[main] SpringCamelContext             INFO  AllowUseOriginalMessage is enabled. If access to the original message is not needed, then its recommended to turn this option off as it may improve performance.
[main] SpringCamelContext             INFO  StreamCaching is not in use. If using streams then its recommended to enable stream caching. See more details at http://camel.apache.org/stream-caching.html
[main] ReflectionServiceFactoryBean   INFO  Creating Service {http://broker.pm.brm.com/}BrokerQouteService from class com.brm.pm.broker.BrokerQouteService
[main] ServerImpl                     INFO  Setting the server's publish address to be http://localhost:1234/broker/loanQouteService
[main] ReflectionServiceFactoryBean   INFO  Creating Service {http://creditburea.pm.brm.com/}CreditScoreService from class com.brm.pm.creditburea.CreditScoreService
[main] ServerImpl                     INFO  Setting the server's publish address to be http://localhost:1234/creditBuruea/CreditScoreService
[main] AggregateProcessor             INFO  Defaulting to MemoryAggregationRepository
[main] AggregateProcessor             INFO  Using CompletionTimeout to trigger after 1000 millis of inactivity.
[main] DefaultCacheManagerFactory     INFO  Creating CacheManager using Ehcache defaults
[main] SpringCamelContext             INFO  Route: route1 started and consuming from: Endpoint[timer://client?period=5000]
[main] Server                         INFO  jetty-8.1.14.v20131031
[main] AbstractConnector              INFO  Started SelectChannelConnector@localhost:1234
[main] SpringCamelContext             INFO  Route: borkerProcessManager started and consuming from: Endpoint[cxf://bean:brokerQouteServiceEndpoint]
[main] SpringCamelContext             INFO  Route: route2 started and consuming from: Endpoint[cxf://bean:creditScoreServiceEndpoint]
[main] SpringCamelContext             INFO  Route: brokerReplyManager started and consuming from: Endpoint[jms://queue:bank.loan.reply]
[main] SpringCamelContext             INFO  Route: Bank1Route started and consuming from: Endpoint[jms://topic:bank.loan.request]
[main] SpringCamelContext             INFO  Route: Bank2Route started and consuming from: Endpoint[jms://topic:bank.loan.request]
[main] SpringCamelContext             INFO  Route: Bank3Route started and consuming from: Endpoint[jms://topic:bank.loan.request]
[main] SpringCamelContext             INFO  Route: Bank4Route started and consuming from: Endpoint[jms://topic:bank.loan.request]
[main] SpringCamelContext             INFO  Route: Bank5Route started and consuming from: Endpoint[jms://topic:bank.loan.request]
[main] SpringCamelContext             INFO  Route: JobSloteRoute started and consuming from: Endpoint[direct://jobslot]
[main] SpringCamelContext             INFO  Total 10 routes, of which 10 is started.
[main] SpringCamelContext             INFO  Apache Camel 2.13.2 (CamelContext: camel-1) started in 1.218 seconds
[Camel (camel-1) thread #7 - timer://client] ReflectionServiceFactoryBean   INFO  Creating Service {http://broker.pm.brm.com/}BrokerQouteService from class com.brm.pm.broker.BrokerQouteService
[qtp2078818709-28] ReflectionServiceFactoryBean   INFO  Creating Service {http://creditburea.pm.brm.com/}CreditScoreService from class com.brm.pm.creditburea.CreditScoreService
result = com.brm.pm.data.LoanQouteReply@1c4a1bda[SSN=1234567890,InterestRate=3.4419281268050064,LoanAmount=100000.0,QouteID=Bank1 : 000004]
result = com.brm.pm.data.LoanQouteReply@4aa46637[SSN=1234567890,InterestRate=3.4042114400702115,LoanAmount=100000.0,QouteID=Bank1 : 000007]
result = com.brm.pm.data.LoanQouteReply@7a5d5033[SSN=1234567890,InterestRate=3.3345451358210463,LoanAmount=100000.0,QouteID=Bank4 : 000011]
result = com.brm.pm.data.LoanQouteReply@52ab6c19[SSN=1234567890,InterestRate=3.2704539831950505,LoanAmount=100000.0,QouteID=Bank3 : 000019]
result = com.brm.pm.data.LoanQouteReply@5b5034e0[SSN=1234567890,InterestRate=3.231147429195609,LoanAmount=100000.0,QouteID=Bank1 : 000023]
result = com.brm.pm.data.LoanQouteReply@63b9ef8c[SSN=1234567890,InterestRate=3.23255684212396,LoanAmount=100000.0,QouteID=Bank3 : 000028]
result = com.brm.pm.data.LoanQouteReply@24e15497[SSN=1234567890,InterestRate=3.3057543515381056,LoanAmount=100000.0,QouteID=Bank2 : 000033]
result = com.brm.pm.data.LoanQouteReply@2699146a[SSN=1234567890,InterestRate=3.238449268535782,LoanAmount=100000.0,QouteID=Bank1 : 000040]
result = com.brm.pm.data.LoanQouteReply@56acf892[SSN=1234567890,InterestRate=3.295696645892292,LoanAmount=100000.0,QouteID=Bank4 : 000045]
result = com.brm.pm.data.LoanQouteReply@27cf35d3[SSN=1234567890,InterestRate=3.490472743892253,LoanAmount=100000.0,QouteID=Bank1 : 000050]
result = com.brm.pm.data.LoanQouteReply@2f971f66[SSN=1234567890,InterestRate=3.2938690658023217,LoanAmount=100000.0,QouteID=Bank3 : 000052]
result = com.brm.pm.data.LoanQouteReply@25c47dd5[SSN=1234567890,InterestRate=3.2339593396144437,LoanAmount=100000.0,QouteID=Bank5 : 000060]
result = com.brm.pm.data.LoanQouteReply@61472b3c[SSN=1234567890,InterestRate=3.571123400154655,LoanAmount=100000.0,QouteID=Bank5 : 000065]
result = com.brm.pm.data.LoanQouteReply@56f93e5b[SSN=1234567890,InterestRate=3.2390674241000164,LoanAmount=100000.0,QouteID=Bank2 : 000066]