1 | /* |
2 | * Copyright 2012 the original author or authors. |
3 | * |
4 | * Licensed under the Apache License, Version 2.0 (the "License"); |
5 | * you may not use this file except in compliance with the License. |
6 | * You may obtain a copy of the License at |
7 | * |
8 | * http://www.apache.org/licenses/LICENSE-2.0 |
9 | * |
10 | * Unless required by applicable law or agreed to in writing, software |
11 | * distributed under the License is distributed on an "AS IS" BASIS, |
12 | * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
13 | * See the License for the specific language governing permissions and |
14 | * limitations under the License. |
15 | */ |
16 | package org.springframework.batch.item.data; |
17 | |
18 | import java.lang.reflect.InvocationTargetException; |
19 | import java.util.ArrayList; |
20 | import java.util.List; |
21 | import java.util.Map; |
22 | |
23 | import org.apache.commons.logging.Log; |
24 | import org.apache.commons.logging.LogFactory; |
25 | import org.springframework.batch.item.ExecutionContext; |
26 | import org.springframework.batch.item.adapter.AbstractMethodInvokingDelegator.InvocationTargetThrowableWrapper; |
27 | import org.springframework.batch.item.adapter.DynamicMethodInvocationException; |
28 | import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader; |
29 | import org.springframework.beans.factory.InitializingBean; |
30 | import org.springframework.data.domain.Page; |
31 | import org.springframework.data.domain.PageRequest; |
32 | import org.springframework.data.domain.Pageable; |
33 | import org.springframework.data.domain.Sort; |
34 | import org.springframework.data.repository.PagingAndSortingRepository; |
35 | import org.springframework.util.Assert; |
36 | import org.springframework.util.ClassUtils; |
37 | import org.springframework.util.MethodInvoker; |
38 | |
39 | /** |
40 | * <p> |
41 | * A {@link org.springframework.batch.item.ItemReader} that reads records utilizing |
42 | * a {@link org.springframework.data.repository.PagingAndSortingRepository}. |
43 | * </p> |
44 | * |
45 | * <p> |
46 | * Performance of the reader is dependent on the repository implementation, however |
47 | * setting a reasonably large page size and matching that to the commit interval should |
48 | * yield better performance. |
49 | * </p> |
50 | * |
51 | * <p> |
52 | * The reader must be configured with a {@link org.springframework.data.repository.PagingAndSortingRepository}, |
53 | * a {@link org.springframework.data.domain.Sort}, and a pageSize greater than 0. |
54 | * </p> |
55 | * |
56 | * <p> |
57 | * This implementation is thread safe between calls to {@link #open(ExecutionContext)}, but remember to use |
58 | * <code>saveState=false</code> if used in a multi-threaded client (no restart available). |
59 | * </p> |
60 | * |
61 | * @author Michael Minella |
62 | * @since 2.2 |
63 | */ |
64 | @SuppressWarnings("rawtypes") |
65 | public class RepositoryItemReader<T> extends AbstractItemCountingItemStreamItemReader<T> implements InitializingBean { |
66 | |
67 | protected Log logger = LogFactory.getLog(getClass()); |
68 | |
69 | private PagingAndSortingRepository repository; |
70 | |
71 | private Sort sort; |
72 | |
73 | private volatile int page = 0; |
74 | |
75 | private int pageSize = 10; |
76 | |
77 | private volatile int current = 0; |
78 | |
79 | private List arguments; |
80 | |
81 | private volatile List<T> results; |
82 | |
83 | private Object lock = new Object(); |
84 | |
85 | private String methodName; |
86 | |
87 | public RepositoryItemReader() { |
88 | setName(ClassUtils.getShortName(RepositoryItemReader.class)); |
89 | } |
90 | |
91 | /** |
92 | * Arguments to be passed to the data providing method. |
93 | * |
94 | * @param arguments list of method arguments to be passed to the repository |
95 | */ |
96 | public void setArguments(List arguments) { |
97 | this.arguments = arguments; |
98 | } |
99 | |
100 | /** |
101 | * Provides ordering of the results so that order is maintained between paged queries |
102 | * |
103 | * @param sorts the fields to sort by and the directions |
104 | */ |
105 | public void setSort(Map<String, Sort.Direction> sorts) { |
106 | this.sort = convertToSort(sorts); |
107 | } |
108 | |
109 | /** |
110 | * @param pageSize The number of items to retrieve per page. |
111 | */ |
112 | public void setPageSize(int pageSize) { |
113 | this.pageSize = pageSize; |
114 | } |
115 | |
116 | /** |
117 | * The {@link org.springframework.data.repository.PagingAndSortingRepository} |
118 | * implementation used to read input from. |
119 | * |
120 | * @param repository underlying repository for input to be read from. |
121 | */ |
122 | public void setRepository(PagingAndSortingRepository repository) { |
123 | this.repository = repository; |
124 | } |
125 | |
126 | /** |
127 | * Specifies what method on the repository to call. This method must take |
128 | * {@link org.springframework.data.domain.Pageable} as the <em>last</em> argument. |
129 | * |
130 | * @param methodName |
131 | */ |
132 | public void setMethodName(String methodName) { |
133 | this.methodName = methodName; |
134 | } |
135 | |
136 | @Override |
137 | public void afterPropertiesSet() throws Exception { |
138 | Assert.state(repository != null, "A PagingAndSortingRepository is required"); |
139 | Assert.state(pageSize > 0, "Page size must be greater than 0"); |
140 | Assert.state(sort != null, "A sort is required"); |
141 | } |
142 | |
143 | @Override |
144 | protected T doRead() throws Exception { |
145 | |
146 | synchronized (lock) { |
147 | if(results == null || current >= results.size()) { |
148 | |
149 | if (logger.isDebugEnabled()) { |
150 | logger.debug("Reading page " + page); |
151 | } |
152 | |
153 | results = doPageRead(); |
154 | |
155 | current = 0; |
156 | page ++; |
157 | |
158 | if(results.size() <= 0) { |
159 | return null; |
160 | } |
161 | } |
162 | |
163 | if(current < results.size()) { |
164 | T curLine = results.get(current); |
165 | current++; |
166 | return curLine; |
167 | } |
168 | else { |
169 | return null; |
170 | } |
171 | } |
172 | } |
173 | |
174 | @Override |
175 | protected void jumpToItem(int itemLastIndex) throws Exception { |
176 | synchronized (lock) { |
177 | page = itemLastIndex / pageSize; |
178 | current = itemLastIndex % pageSize; |
179 | |
180 | results = doPageRead(); |
181 | page++; |
182 | } |
183 | } |
184 | |
185 | /** |
186 | * Performs the actual reading of a page via the repository. |
187 | * Available for overriding as needed. |
188 | * |
189 | * @return the list of items that make up the page |
190 | * @throws Exception |
191 | */ |
192 | @SuppressWarnings("unchecked") |
193 | protected List<T> doPageRead() throws Exception { |
194 | Pageable pageRequest = new PageRequest(page, pageSize, sort); |
195 | |
196 | MethodInvoker invoker = createMethodInvoker(repository, methodName); |
197 | |
198 | List parameters = new ArrayList(); |
199 | |
200 | if(arguments != null && arguments.size() > 0) { |
201 | parameters.addAll(arguments); |
202 | } |
203 | |
204 | parameters.add(pageRequest); |
205 | |
206 | invoker.setArguments(parameters.toArray()); |
207 | |
208 | Page curPage = (Page) doInvoke(invoker); |
209 | |
210 | return curPage.getContent(); |
211 | } |
212 | |
213 | @Override |
214 | protected void doOpen() throws Exception { |
215 | } |
216 | |
217 | @Override |
218 | protected void doClose() throws Exception { |
219 | } |
220 | |
221 | private Sort convertToSort(Map<String, Sort.Direction> sorts) { |
222 | List<Sort.Order> sortValues = new ArrayList<Sort.Order>(); |
223 | |
224 | for (Map.Entry<String, Sort.Direction> curSort : sorts.entrySet()) { |
225 | sortValues.add(new Sort.Order(curSort.getValue(), curSort.getKey())); |
226 | } |
227 | |
228 | return new Sort(sortValues); |
229 | } |
230 | |
231 | private Object doInvoke(MethodInvoker invoker) throws Exception{ |
232 | try { |
233 | invoker.prepare(); |
234 | } |
235 | catch (ClassNotFoundException e) { |
236 | throw new DynamicMethodInvocationException(e); |
237 | } |
238 | catch (NoSuchMethodException e) { |
239 | throw new DynamicMethodInvocationException(e); |
240 | } |
241 | |
242 | try { |
243 | return invoker.invoke(); |
244 | } |
245 | catch (InvocationTargetException e) { |
246 | if (e.getCause() instanceof Exception) { |
247 | throw (Exception) e.getCause(); |
248 | } |
249 | else { |
250 | throw new InvocationTargetThrowableWrapper(e.getCause()); |
251 | } |
252 | } |
253 | catch (IllegalAccessException e) { |
254 | throw new DynamicMethodInvocationException(e); |
255 | } |
256 | } |
257 | |
258 | private MethodInvoker createMethodInvoker(Object targetObject, String targetMethod) { |
259 | MethodInvoker invoker = new MethodInvoker(); |
260 | invoker.setTargetObject(targetObject); |
261 | invoker.setTargetMethod(targetMethod); |
262 | return invoker; |
263 | } |
264 | } |