1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package org.codelibs.fess.helper;
17
18 import java.util.ArrayList;
19 import java.util.Collections;
20 import java.util.HashMap;
21 import java.util.List;
22 import java.util.Map;
23
24 import javax.annotation.Resource;
25
26 import org.codelibs.core.lang.StringUtil;
27 import org.codelibs.fess.Constants;
28 import org.codelibs.fess.app.service.DataConfigService;
29 import org.codelibs.fess.app.service.FailureUrlService;
30 import org.codelibs.fess.ds.DataStore;
31 import org.codelibs.fess.ds.DataStoreFactory;
32 import org.codelibs.fess.ds.IndexUpdateCallback;
33 import org.codelibs.fess.es.client.FessEsClient;
34 import org.codelibs.fess.es.config.exentity.DataConfig;
35 import org.codelibs.fess.mylasta.direction.FessConfig;
36 import org.codelibs.fess.util.ComponentUtil;
37 import org.elasticsearch.index.query.QueryBuilder;
38 import org.elasticsearch.index.query.QueryBuilders;
39 import org.slf4j.Logger;
40 import org.slf4j.LoggerFactory;
41
42 public class DataIndexHelper {
43
44 private static final Logger logger = LoggerFactory.getLogger(DataIndexHelper.class);
45
46 private static final String DELETE_OLD_DOCS = "delete_old_docs";
47
48 @Resource
49 public DataConfigService dataConfigService;
50
51 @Resource
52 protected CrawlingConfigHelper crawlingConfigHelper;
53
54 public long crawlingExecutionInterval = Constants.DEFAULT_CRAWLING_EXECUTION_INTERVAL;
55
56 public int crawlerPriority = Thread.NORM_PRIORITY;
57
58 private final List<DataCrawlingThread> dataCrawlingThreadList = Collections.synchronizedList(new ArrayList<DataCrawlingThread>());
59
60 public void crawl(final String sessionId) {
61 final List<DataConfig> configList = dataConfigService.getAllDataConfigList();
62
63 if (configList.isEmpty()) {
64
65 if (logger.isInfoEnabled()) {
66 logger.info("No crawling target data.");
67 }
68 return;
69 }
70
71 doCrawl(sessionId, configList);
72 }
73
74 public void crawl(final String sessionId, final List<String> configIdList) {
75 final List<DataConfig> configList = dataConfigService.getDataConfigListByIds(configIdList);
76
77 if (configList.isEmpty()) {
78
79 if (logger.isInfoEnabled()) {
80 logger.info("No crawling target urls.");
81 }
82 return;
83 }
84
85 doCrawl(sessionId, configList);
86 }
87
88 protected void doCrawl(final String sessionId, final List<DataConfig> configList) {
89 final int multiprocessCrawlingCount = ComponentUtil.getFessConfig().getCrawlingThreadCount();
90
91 final long startTime = System.currentTimeMillis();
92
93 final IndexUpdateCallback indexUpdateCallback = ComponentUtil.getComponent(IndexUpdateCallback.class);
94
95 final List<String> sessionIdList = new ArrayList<>();
96 final Map<String, String> initParamMap = new HashMap<>();
97 dataCrawlingThreadList.clear();
98 final List<String> dataCrawlingThreadStatusList = new ArrayList<>();
99 for (final DataConfig dataConfig : configList) {
100 final String sid = crawlingConfigHelper.store(sessionId, dataConfig);
101 sessionIdList.add(sid);
102
103 initParamMap.put(Constants.SESSION_ID, sessionId);
104 initParamMap.put(Constants.CRAWLING_INFO_ID, sid);
105
106 final DataCrawlingThread dataCrawlingThread = new DataCrawlingThread(dataConfig, indexUpdateCallback, initParamMap);
107 dataCrawlingThread.setPriority(crawlerPriority);
108 dataCrawlingThread.setName(sid);
109 dataCrawlingThread.setDaemon(true);
110
111 dataCrawlingThreadList.add(dataCrawlingThread);
112 dataCrawlingThreadStatusList.add(Constants.READY);
113
114 }
115
116 final SystemHelper systemHelper = ComponentUtil.getSystemHelper();
117
118 int startedCrawlerNum = 0;
119 int activeCrawlerNum = 0;
120 while (startedCrawlerNum < dataCrawlingThreadList.size()) {
121
122 if (systemHelper.isForceStop()) {
123 for (final DataCrawlingThread crawlerThread : dataCrawlingThreadList) {
124 crawlerThread.stopCrawling();
125 }
126 break;
127 }
128
129 if (activeCrawlerNum < multiprocessCrawlingCount) {
130
131 dataCrawlingThreadList.get(startedCrawlerNum).start();
132 dataCrawlingThreadStatusList.set(startedCrawlerNum, Constants.RUNNING);
133 startedCrawlerNum++;
134 activeCrawlerNum++;
135 try {
136 Thread.sleep(crawlingExecutionInterval);
137 } catch (final InterruptedException e) {
138 if (logger.isDebugEnabled()) {
139 logger.debug("Interrupted.", e);
140 }
141 }
142 continue;
143 }
144
145
146 for (int i = 0; i < startedCrawlerNum; i++) {
147 if (!dataCrawlingThreadList.get(i).isRunning() && dataCrawlingThreadStatusList.get(i).equals(Constants.RUNNING)) {
148 dataCrawlingThreadList.get(i).awaitTermination();
149 dataCrawlingThreadStatusList.set(i, Constants.DONE);
150 activeCrawlerNum--;
151 }
152 }
153 try {
154 Thread.sleep(crawlingExecutionInterval);
155 } catch (final InterruptedException e) {
156 if (logger.isDebugEnabled()) {
157 logger.debug("Interrupted.", e);
158 }
159 }
160 }
161
162 boolean finishedAll = false;
163 while (!finishedAll) {
164 finishedAll = true;
165 for (int i = 0; i < dataCrawlingThreadList.size(); i++) {
166 dataCrawlingThreadList.get(i).awaitTermination(crawlingExecutionInterval);
167 if (!dataCrawlingThreadList.get(i).isRunning() && dataCrawlingThreadStatusList.get(i).equals(Constants.RUNNING)) {
168 dataCrawlingThreadStatusList.set(i, Constants.DONE);
169 }
170 if (!dataCrawlingThreadStatusList.get(i).equals(Constants.DONE)) {
171 finishedAll = false;
172 }
173 }
174 }
175 dataCrawlingThreadList.clear();
176 dataCrawlingThreadStatusList.clear();
177
178
179 final CrawlingInfoHelper crawlingInfoHelper = ComponentUtil.getCrawlingInfoHelper();
180
181 final long execTime = System.currentTimeMillis() - startTime;
182 crawlingInfoHelper.putToInfoMap(Constants.DATA_CRAWLING_EXEC_TIME, Long.toString(execTime));
183 if (logger.isInfoEnabled()) {
184 logger.info("[EXEC TIME] crawling time: " + execTime + "ms");
185 }
186
187 crawlingInfoHelper.putToInfoMap(Constants.DATA_INDEX_EXEC_TIME, Long.toString(indexUpdateCallback.getExecuteTime()));
188 crawlingInfoHelper.putToInfoMap(Constants.DATA_INDEX_SIZE, Long.toString(indexUpdateCallback.getDocumentSize()));
189
190 for (final String sid : sessionIdList) {
191
192 crawlingConfigHelper.remove(sid);
193 }
194
195 }
196
197 protected static class DataCrawlingThread extends Thread {
198
199 private final DataConfig dataConfig;
200
201 private final IndexUpdateCallback indexUpdateCallback;
202
203 private final Map<String, String> initParamMap;
204
205 protected boolean finished = false;
206
207 protected boolean running = false;
208
209 private DataStore dataStore;
210
211 protected DataCrawlingThread(final DataConfig dataConfig, final IndexUpdateCallback indexUpdateCallback,
212 final Map<String, String> initParamMap) {
213 this.dataConfig = dataConfig;
214 this.indexUpdateCallback = indexUpdateCallback;
215 this.initParamMap = initParamMap;
216 }
217
218 @Override
219 public void run() {
220 running = true;
221 try {
222 process();
223 } finally {
224 running = false;
225 finished = true;
226 }
227 }
228
229 protected void process() {
230 final DataStoreFactory dataStoreFactory = ComponentUtil.getDataStoreFactory();
231 dataStore = dataStoreFactory.getDataStore(dataConfig.getHandlerName());
232 if (dataStore == null) {
233 logger.error("DataStore(" + dataConfig.getHandlerName() + ") is not found.");
234 } else {
235 try {
236 dataStore.store(dataConfig, indexUpdateCallback, initParamMap);
237 } catch (final Throwable e) {
238 logger.error("Failed to process a data crawling: " + dataConfig.getName(), e);
239 ComponentUtil.getComponent(FailureUrlService.class).store(dataConfig, e.getClass().getCanonicalName(),
240 dataConfig.getConfigId() + ":" + dataConfig.getName(), e);
241 } finally {
242 indexUpdateCallback.commit();
243 deleteOldDocs();
244 }
245 }
246 }
247
248 private void deleteOldDocs() {
249 if (Constants.FALSE.equals(initParamMap.get(DELETE_OLD_DOCS))) {
250 return;
251 }
252 final String sessionId = initParamMap.get(Constants.SESSION_ID);
253 if (StringUtil.isBlank(sessionId)) {
254 logger.warn("Invalid sessionId at " + dataConfig);
255 return;
256 }
257 final FessConfig fessConfig = ComponentUtil.getFessConfig();
258 final QueryBuilder queryBuilder =
259 QueryBuilders
260 .boolQuery()
261 .must(QueryBuilders.termQuery(fessConfig.getIndexFieldConfigId(), dataConfig.getConfigId()))
262 .must(QueryBuilders.boolQuery().mustNot(QueryBuilders.rangeQuery(fessConfig.getIndexFieldExpires()).gt("now"))
263 .mustNot(QueryBuilders.existsQuery(fessConfig.getIndexFieldExpires())))
264 .mustNot(QueryBuilders.termQuery(fessConfig.getIndexFieldSegment(), sessionId));
265 try {
266 final FessEsClient fessEsClient = ComponentUtil.getFessEsClient();
267 final String index = fessConfig.getIndexDocumentUpdateIndex();
268 fessEsClient.admin().indices().prepareRefresh(index).execute().actionGet();
269 final int numOfDeleted = fessEsClient.deleteByQuery(index, fessConfig.getIndexDocumentType(), queryBuilder);
270 logger.info("Deleted {} old docs.", numOfDeleted);
271 } catch (final Exception e) {
272 logger.error("Could not delete old docs at " + dataConfig, e);
273 }
274 }
275
276 public boolean isFinished() {
277 return finished;
278 }
279
280 public void stopCrawling() {
281 if (dataStore != null) {
282 dataStore.stop();
283 }
284 }
285
286 public String getCrawlingInfoId() {
287 return initParamMap.get(Constants.CRAWLING_INFO_ID);
288 }
289
290 public boolean isRunning() {
291 return running;
292 }
293
294 public void awaitTermination() {
295 try {
296 join();
297 } catch (final InterruptedException e) {
298 if (logger.isDebugEnabled()) {
299 logger.debug("Interrupted.", e);
300 }
301 }
302 }
303
304 public void awaitTermination(final long mills) {
305 try {
306 join(mills);
307 } catch (final InterruptedException e) {
308 if (logger.isDebugEnabled()) {
309 logger.debug("Interrupted.", e);
310 }
311 }
312 }
313 }
314 }