Spring/spring-integration-extensions
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INTEXT-43 replaced ConnectionFactory with ServiceFactory

Browse Source 
* remove trailing spaces from SplunkEventTests
* remove trailing spaces from README.md
This commit is contained in:
David Turanski
2013-02-05 18:23:00 -05:00
committed by Gunnar Hillert
parent 9eaed7e08a
commit 885e2ebc11
28 changed files with 183 additions and 805 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/config/xml/SplunkInboundChannelAdapterParser.java
View File

@@ -23,9 +23,8 @@ import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.splunk.inbound.SplunkPollingChannelAdapter;
import org.springframework.integration.splunk.support.ConnectionFactoryFactoryBean;
import org.springframework.integration.splunk.support.SplunkConnectionFactory;
import org.springframework.integration.splunk.support.SplunkDataReader;
import org.springframework.integration.splunk.support.SplunkServiceFactory;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
@@ -61,18 +60,14 @@ public class SplunkInboundChannelAdapterParser extends AbstractPollingInboundCha
IntegrationNamespaceUtils.setValueIfAttributeDefined(splunkDataReaderBuilder, element, "earliest-time");
IntegrationNamespaceUtils.setValueIfAttributeDefined(splunkDataReaderBuilder, element, "latest-time");
BeanDefinitionBuilder connectionFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(SplunkConnectionFactory.class);
BeanDefinitionBuilder serviceFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(SplunkServiceFactory.class);
String splunkServerBeanName = element.getAttribute("splunk-server-ref");
if (StringUtils.hasText(splunkServerBeanName)) {
connectionFactoryBuilder.addConstructorArgReference(splunkServerBeanName);
serviceFactoryBuilder.addConstructorArgReference(splunkServerBeanName);
}
BeanDefinitionBuilder connectionFactoryFactoryBeanBuilder = BeanDefinitionBuilder.genericBeanDefinition(ConnectionFactoryFactoryBean.class);
connectionFactoryFactoryBeanBuilder.addConstructorArgValue(connectionFactoryBuilder.getBeanDefinition());
connectionFactoryFactoryBeanBuilder.addConstructorArgValue(element.getAttribute("pool-server-connection"));
splunkDataReaderBuilder.addConstructorArgValue(connectionFactoryFactoryBeanBuilder.getBeanDefinition());
splunkDataReaderBuilder.addConstructorArgValue(serviceFactoryBuilder.getBeanDefinition());
String channelAdapterId = this.resolveId(element, splunkPollingChannelAdapterBuilder.getRawBeanDefinition(),
parserContext);

34
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/config/xml/SplunkOutboundChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -23,10 +23,9 @@ import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.splunk.outbound.SplunkOutboundChannelAdapter;
import org.springframework.integration.splunk.support.ConnectionFactoryFactoryBean;
import org.springframework.integration.splunk.support.SplunkArgsFactoryBean;
import org.springframework.integration.splunk.support.SplunkConnectionFactory;
import org.springframework.integration.splunk.support.SplunkIndexWriter;
import org.springframework.integration.splunk.support.SplunkServiceFactory;
import org.springframework.integration.splunk.support.SplunkSubmitWriter;
import org.springframework.integration.splunk.support.SplunkTcpWriter;
import org.springframework.util.StringUtils;
@@ -59,27 +58,24 @@ public class SplunkOutboundChannelAdapterParser extends AbstractOutboundChannelA
BeanDefinitionBuilder splunkOutboundChannelAdapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(SplunkOutboundChannelAdapter.class);
BeanDefinitionBuilder splunkExecutorBuilder = SplunkParserUtils.getSplunkExecutorBuilder(element, parserContext);
BeanDefinitionBuilder argsBuilder = BeanDefinitionBuilder.genericBeanDefinition(SplunkArgsFactoryBean.class);
IntegrationNamespaceUtils.setValueIfAttributeDefined(argsBuilder, element, "source-type");
IntegrationNamespaceUtils.setValueIfAttributeDefined(argsBuilder, element, "source");
IntegrationNamespaceUtils.setValueIfAttributeDefined(argsBuilder, element, "host");
IntegrationNamespaceUtils.setValueIfAttributeDefined(argsBuilder, element, "host-regex");
BeanDefinitionBuilder connectionFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(SplunkConnectionFactory.class);
String splunkServerBeanName = element.getAttribute("splunk-server-ref");
BeanDefinitionBuilder serviceFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(SplunkServiceFactory.class);
String splunkServerBeanName = element.getAttribute("splunk-server-ref");
if (StringUtils.hasText(splunkServerBeanName)) {
connectionFactoryBuilder.addConstructorArgReference(splunkServerBeanName);
serviceFactoryBuilder.addConstructorArgReference(splunkServerBeanName);
}
BeanDefinitionBuilder connectionFactoryFactoryBeanBuilder = BeanDefinitionBuilder.genericBeanDefinition(ConnectionFactoryFactoryBean.class);
connectionFactoryFactoryBeanBuilder.addConstructorArgValue(connectionFactoryBuilder.getBeanDefinition());
connectionFactoryFactoryBeanBuilder.addConstructorArgValue(element.getAttribute("pool-server-connection"));
BeanDefinitionBuilder dataWriterBuilder = parseDataWriter(element, parserContext);
dataWriterBuilder.addConstructorArgValue(connectionFactoryFactoryBeanBuilder.getBeanDefinition());
dataWriterBuilder.addConstructorArgValue(serviceFactoryBuilder.getBeanDefinition());
dataWriterBuilder.addConstructorArgValue(argsBuilder.getBeanDefinition());
String channelAdapterId = this.resolveId(element, splunkOutboundChannelAdapterBuilder.getRawBeanDefinition(),
parserContext);
String splunkExecutorBeanName = channelAdapterId + ".splunkExecutor";
@@ -99,7 +95,7 @@ public class SplunkOutboundChannelAdapterParser extends AbstractOutboundChannelA
return splunkOutboundChannelAdapterBuilder.getBeanDefinition();
}
private BeanDefinitionBuilder parseDataWriter(Element element, ParserContext parserContext) {
BeanDefinitionBuilder dataWriterBuilder = null;
Element dataWriter = null;
@@ -118,11 +114,11 @@ public class SplunkOutboundChannelAdapterParser extends AbstractOutboundChannelA
dataWriter = DomUtils.getChildElementByTagName(element, "tcp-writer");
dataWriterBuilder = BeanDefinitionBuilder.genericBeanDefinition(SplunkTcpWriter.class);
IntegrationNamespaceUtils.setValueIfAttributeDefined(dataWriterBuilder, dataWriter, "port");
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(dataWriterBuilder, element, "auto-startup");
return dataWriterBuilder;
}

34
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/core/Connection.java
View File

@@ -1,34 +0,0 @@
/*
* Copyright 2011-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.splunk.core;
/**
* Connection to Splunk service
*
* @author Jarred Li
* @since 1.0
*
*/
public interface Connection<T> {
T getTarget();
void close();
boolean isOpen();
}

29
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/core/ConnectionFactory.java
View File

@@ -1,29 +0,0 @@
/*
* Copyright 2011-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.splunk.core;
/**
* Factory pattern to create <code>Connection</code>
*
* @author Jarred Li
* @since 1.0
*
*/
public interface ConnectionFactory<T> {
Connection<T> getConnection() throws Exception;
}

2
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/core/DataReader.java
View File

@@ -27,6 +27,6 @@ import org.springframework.integration.splunk.event.SplunkEvent;
*/
public interface DataReader {
List<SplunkEvent> search() throws Exception;
List<SplunkEvent> read() throws Exception;
}

23
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/core/ServiceFactory.java Normal file
View File

@@ -0,0 +1,23 @@
/*
* Copyright 2002-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
* specific language governing permissions and limitations under the License.
*/
package org.springframework.integration.splunk.core;
import com.splunk.Service;
/**
* @author David Turanski
*
*/
public interface ServiceFactory {
public Service getService();
}

2
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/outbound/SplunkOutboundChannelAdapter.java
View File

@@ -58,7 +58,7 @@ public class SplunkOutboundChannelAdapter extends AbstractReplyProducingMessageH
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
final Object result;
result = this.splunkExecutor.executeOutboundOperation(requestMessage);
result = this.splunkExecutor.write(requestMessage);
if (result == null || !producesReply) {
return null;
}

21
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/support/AbstractSplunkDataWriter.java
View File

@@ -24,9 +24,8 @@ import java.net.Socket;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.splunk.core.Connection;
import org.springframework.integration.splunk.core.ConnectionFactory;
import org.springframework.integration.splunk.core.DataWriter;
import org.springframework.integration.splunk.core.ServiceFactory;
import org.springframework.integration.splunk.event.SplunkEvent;
import org.springframework.util.Assert;
@@ -45,8 +44,6 @@ public abstract class AbstractSplunkDataWriter implements DataWriter, SmartLifec
protected final Log logger = LogFactory.getLog(getClass());
protected ConnectionFactory<Service> connectionFactory;
protected Socket socket;
protected Service service;
@@ -58,12 +55,14 @@ public abstract class AbstractSplunkDataWriter implements DataWriter, SmartLifec
private int phase;
private boolean autoStartup = true;
private final ServiceFactory serviceFactory;
protected AbstractSplunkDataWriter(ConnectionFactory<Service> connectionFactory, Args args) {
Assert.notNull(connectionFactory,"connectionFactory cannot be null");
this.connectionFactory = connectionFactory;
protected AbstractSplunkDataWriter(ServiceFactory serviceFactory, Args args) {
Assert.notNull(serviceFactory,"service factory cannot be null");
this.serviceFactory = serviceFactory;
Assert.notNull(args, "args cannot be null");
this.args = args;
@@ -96,10 +95,8 @@ public abstract class AbstractSplunkDataWriter implements DataWriter, SmartLifec
*/
public synchronized void start() {
try {
Connection<Service> connection = connectionFactory.getConnection();
this.service = connection.getTarget();
socket = createSocket(service);
service = serviceFactory.getService();
socket = createSocket(service);
} catch (Exception e) {
throw new RuntimeException(e);
@@ -118,8 +115,6 @@ public abstract class AbstractSplunkDataWriter implements DataWriter, SmartLifec
if (socket != null) {
socket.close();
}
connectionFactory.getConnection().close();
} catch (Exception e) {
throw new RuntimeException(e);
}

62
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/support/ConnectionFactoryFactoryBean.java
View File

@@ -1,62 +0,0 @@
/*
* Copyright 2011-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.splunk.support;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.integration.splunk.core.ConnectionFactory;
/**
* Factory bean to create <code>ConnectionFactory</code>.
*
* @author Jarred Li
* @since 1.0
*
*/
public class ConnectionFactoryFactoryBean<T> implements FactoryBean<ConnectionFactory<T>> {
private final ConnectionFactory<T> connectionFactory;
public ConnectionFactoryFactoryBean(ConnectionFactory<T> cf, boolean usePool) {
if (usePool) {
this.connectionFactory = new PoolingConnectionFactory<T>(cf);
}
else {
this.connectionFactory = cf;
}
}
/* (non-Javadoc)
* @see org.springframework.beans.factory.FactoryBean#getObject()
*/
public ConnectionFactory<T> getObject() throws Exception {
return this.connectionFactory;
}
/* (non-Javadoc)
* @see org.springframework.beans.factory.FactoryBean#getObjectType()
*/
public Class<?> getObjectType() {
return connectionFactory.getClass();
}
/* (non-Javadoc)
* @see org.springframework.beans.factory.FactoryBean#isSingleton()
*/
public boolean isSingleton() {
return true;
}
}

134
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/support/PoolingConnectionFactory.java
View File

@@ -1,134 +0,0 @@
/*
* Copyright 2011-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.splunk.support;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.pool.BasePoolableObjectFactory;
import org.apache.commons.pool.ObjectPool;
import org.apache.commons.pool.impl.GenericObjectPool;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.integration.splunk.core.Connection;
import org.springframework.integration.splunk.core.ConnectionFactory;
/**
* Pooling ConnectionFactory to pool <code>Connection</code> with Apache Commons Pool.
*
* @author Jarred Li
* @since 1.0
*
*/
public class PoolingConnectionFactory<T> implements ConnectionFactory<T>, DisposableBean {
private final Log log = LogFactory.getLog(this.getClass());
private final ConnectionFactory<T> connectionFactory;
private ObjectPool<Connection<T>> pool;
public PoolingConnectionFactory(ConnectionFactory<T> f) {
this.connectionFactory = f;
this.pool = new GenericObjectPool<Connection<T>>(new ConnectionPoolableObjectFactory());
}
/* (non-Javadoc)
* @see org.springframework.integration.splunk.core.ServiceFactory#getService()
*/
public Connection<T> getConnection() throws Exception {
return new PooledConnection(this.pool.borrowObject());
}
/* (non-Javadoc)
* @see org.springframework.beans.factory.DisposableBean#destroy()
*/
public void destroy() throws Exception {
pool.clear();
pool.close();
}
class ConnectionPoolableObjectFactory extends BasePoolableObjectFactory<Connection<T>> {
/* (non-Javadoc)
* @see org.apache.commons.pool.BasePoolableObjectFactory#makeObject()
*/
@Override
public Connection<T> makeObject() throws Exception {
return connectionFactory.getConnection();
}
@Override
public void destroyObject(Connection<T> obj) throws Exception {
obj.close();
}
/**
* Whether the object is valid or not.
*
* @param obj object to be validated
* @return <tt>true</tt>
*/
public boolean validateObject(Connection<T> obj) {
return obj.isOpen();
}
/**
* activate the object
*
* @param obj ignored
*/
public void activateObject(Connection<T> obj) throws Exception {
obj.isOpen();
}
}
class PooledConnection implements Connection<T> {
private Connection<T> connection;
public PooledConnection(Connection<T> con) {
this.connection = con;
}
/* (non-Javadoc)l
* @see org.springframework.integration.splunk.core.IService#close()
*/
public void close() {
try {
pool.returnObject(connection);
} catch (Exception e) {
log.warn("failed to return pooled object", e);
}
}
/* (non-Javadoc)
* @see org.springframework.integration.splunk.core.IService#isOpen()
*/
public boolean isOpen() {
return connection.isOpen();
}
/* (non-Javadoc)
* @see org.springframework.integration.splunk.core.IService#getService()
*/
public T getTarget() {
return connection.getTarget();
}
}
}

49
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/support/SplunkConnectionFactory.java
View File

@@ -1,49 +0,0 @@
/*
* Copyright 2011-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.splunk.support;
import org.springframework.integration.splunk.core.Connection;
import org.springframework.integration.splunk.core.ConnectionFactory;
import com.splunk.Service;
/**
* Factory to create Splunk connection.
*
* @author Jarred Li
* @author David Turanski
* @since 1.0
*
*/
public class SplunkConnectionFactory implements ConnectionFactory<Service> {
private SplunkServer splunkServer;
private SplunkConnection connection;
public SplunkConnectionFactory(SplunkServer server) {
this.splunkServer = server;
}
/* (non-Javadoc)
* @see org.springframework.integration.splunk.core.ServiceFactory#getService()
*/
public synchronized Connection<Service> getConnection() throws Exception {
if (connection == null || !connection.isOpen()) {
connection = new SplunkConnection(splunkServer);
}
return connection;
}
}

82
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/support/SplunkDataReader.java
View File

@@ -27,9 +27,8 @@ import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.splunk.core.Connection;
import org.springframework.integration.splunk.core.ConnectionFactory;
import org.springframework.integration.splunk.core.DataReader;
import org.springframework.integration.splunk.core.ServiceFactory;
import org.springframework.integration.splunk.event.SplunkEvent;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -65,8 +64,6 @@ public class SplunkDataReader implements DataReader, InitializingBean {
private static final Log logger = LogFactory.getLog(SplunkDataReader.class);
private ConnectionFactory<Service> connectionFactory;
private SearchMode mode;
private int count = 0;
@@ -89,8 +86,10 @@ public class SplunkDataReader implements DataReader, InitializingBean {
private transient Calendar lastSuccessfulReadTime;
public SplunkDataReader(ConnectionFactory<Service> f) {
this.connectionFactory = f;
private final ServiceFactory serviceFactory;
public SplunkDataReader(ServiceFactory serviceFactory) {
this.serviceFactory = serviceFactory;
}
public void setSearch(String searchStr) {
@@ -176,7 +175,7 @@ public class SplunkDataReader implements DataReader, InitializingBean {
return app;
}
public List<SplunkEvent> search() throws Exception {
public List<SplunkEvent> read() throws Exception {
logger.debug("mode:" + mode);
switch (mode) {
case SAVEDSEARCH: {
@@ -217,7 +216,6 @@ public class SplunkDataReader implements DataReader, InitializingBean {
return result;
}
/**
* get earliest time for realtime search
*
@@ -253,12 +251,10 @@ public class SplunkDataReader implements DataReader, InitializingBean {
String lTime = null;
if (StringUtils.hasText(latestTime)) {
lTime = latestTime;
}
else {
} else {
if (realtime) {
lTime = "rt";
}
else {
} else {
DateFormat df = new SimpleDateFormat(DATE_FORMAT);
lTime = df.format(startTime.getTime());
}
@@ -271,18 +267,15 @@ public class SplunkDataReader implements DataReader, InitializingBean {
if (lastSuccessfulReadTime == null) {
eTime = initEarliestTime;
}
else {
} else {
if (StringUtils.hasText(earliestTime)) {
eTime = earliestTime;
}
else {
} else {
String calculatedEarliestTime = calculateEarliestTime(startTime, realtime);
if (calculatedEarliestTime != null) {
if (realtime) {
eTime = "rt" + calculatedEarliestTime;
}
else {
} else {
eTime = calculatedEarliestTime;
}
}
@@ -291,18 +284,14 @@ public class SplunkDataReader implements DataReader, InitializingBean {
return eTime;
}
private List<SplunkEvent> runQuery(Args queryArgs) throws Exception {
Connection<Service> connection = connectionFactory.getConnection();
try {
Job job = connection.getTarget().getJobs().create(search, queryArgs);
while (!job.isDone()) {
Thread.sleep(2000);
}
return extractData(job);
} finally {
connection.close();
Service service = serviceFactory.getService();
Job job = service.getJobs().create(search, queryArgs);
while (!job.isDone()) {
Thread.sleep(2000);
}
return extractData(job);
}
private List<SplunkEvent> blockingSearch() throws Exception {
@@ -317,7 +306,6 @@ public class SplunkDataReader implements DataReader, InitializingBean {
return data;
}
private List<SplunkEvent> nonBlockingSearch() throws Exception {
logger.debug("non block search start");
@@ -331,7 +319,6 @@ public class SplunkDataReader implements DataReader, InitializingBean {
return data;
}
/**
* @return
* @throws Exception
@@ -364,21 +351,16 @@ public class SplunkDataReader implements DataReader, InitializingBean {
populateArgs(queryArgs, startTime, false);
queryArgs.put("output_mode", "xml");
Connection<Service> connection = connectionFactory.getConnection();
try {
InputStream os = connection.getTarget().export(search, queryArgs);
ResultsReaderXml resultsReader = new ResultsReaderXml(os);
while ((data = resultsReader.getNextEvent()) != null) {
splunkData = new SplunkEvent(data);
result.add(splunkData);
}
return result;
} finally {
connection.close();
Service service = serviceFactory.getService();
InputStream os = service.export(search, queryArgs);
ResultsReaderXml resultsReader = new ResultsReaderXml(os);
while ((data = resultsReader.getNextEvent()) != null) {
splunkData = new SplunkEvent(data);
result.add(splunkData);
}
return result;
}
private List<SplunkEvent> savedSearch() throws Exception {
logger.debug("saved search start");
@@ -392,13 +374,14 @@ public class SplunkDataReader implements DataReader, InitializingBean {
}
Calendar startTime = Calendar.getInstance();
Connection<Service> connection = connectionFactory.getConnection();
try {
SavedSearch search = null;
Job job = null;
String latestTime = getLatestTime(startTime, false);
String earliestTime = getEarliestTime(startTime, false);
SavedSearchCollection savedSearches = connection.getTarget().getSavedSearches(queryArgs);
Service service = serviceFactory.getService();
SavedSearchCollection savedSearches = service.getSavedSearches(queryArgs);
for (SavedSearch s : savedSearches.values()) {
if (s.getName().equals(savedSearch)) {
search = s;
@@ -417,9 +400,7 @@ public class SplunkDataReader implements DataReader, InitializingBean {
List<SplunkEvent> data = extractData(job);
this.lastSuccessfulReadTime = startTime;
return data;
} finally {
connection.close();
}
}
private List<SplunkEvent> extractData(Job job) throws Exception {
@@ -440,8 +421,7 @@ public class SplunkDataReader implements DataReader, InitializingBean {
splunkData = new SplunkEvent(data);
result.add(splunkData);
}
}
else {
} else {
int offset = 0;
while (offset < total) {
InputStream stream = null;
@@ -465,6 +445,4 @@ public class SplunkDataReader implements DataReader, InitializingBean {
Assert.notNull(initEarliestTime, "initial earliest time can not be null");
}
}

21
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/support/SplunkExecutor.java
View File

@@ -19,7 +19,6 @@ import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.MessagingException;
@@ -31,31 +30,21 @@ import org.springframework.integration.splunk.event.SplunkEvent;
* Bundles common core logic for the Splunk components.
*
* @author Jarred Li
* @author David Turanski
* @since 1.0
*
*/
public class SplunkExecutor implements InitializingBean {
public class SplunkExecutor {
private static final Log logger = LogFactory.getLog(SplunkExecutor.class);
private DataReader reader;
private DataWriter writer;
public SplunkExecutor() {
}
/**
* Verifies and sets the parameters. E.g. initializes the to be used
*/
public void afterPropertiesSet() {
}
/**
* Executes the outbound Splunk Operation.
*
*/
public Object executeOutboundOperation(final Message<?> message) {
public Object write(final Message<?> message) {
try {
SplunkEvent payload = (SplunkEvent) message.getPayload();
writer.write(payload);
@@ -68,7 +57,7 @@ public class SplunkExecutor implements InitializingBean {
}
public void handleMessage(final Message<?> message) {
executeOutboundOperation(message);
write(message);
}
/**
@@ -78,7 +67,7 @@ public class SplunkExecutor implements InitializingBean {
logger.debug("poll start:");
List<SplunkEvent> queryData = null;
try {
queryData = reader.search();
queryData = reader.read();
} catch (Exception e) {
String errorMsg = "search Splunk data failed";
logger.warn(errorMsg, e);

6
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/support/SplunkIndexWriter.java
View File

@@ -15,7 +15,7 @@ package org.springframework.integration.splunk.support;
import java.io.IOException;
import java.net.Socket;
import org.springframework.integration.splunk.core.ConnectionFactory;
import org.springframework.integration.splunk.core.ServiceFactory;
import org.springframework.util.Assert;
import com.splunk.Args;
@@ -41,8 +41,8 @@ public class SplunkIndexWriter extends AbstractSplunkDataWriter {
* @param connectionFactory
* @param args
*/
public SplunkIndexWriter(ConnectionFactory<Service> connectionFactory, Args args) {
super(connectionFactory, args);
public SplunkIndexWriter(ServiceFactory serviceFactory, Args args) {
super(serviceFactory, args);
}
/* (non-Javadoc)

76
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/support/SplunkConnection.java → spring-integration-splunk/src/main/java/org/springframework/integration/splunk/support/SplunkServiceFactory.java
View File

@@ -1,17 +1,14 @@
/*
* Copyright 2011-2012 the original author or authors.
* Copyright 2002-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
* specific language governing permissions and limitations under the License.
*/
package org.springframework.integration.splunk.support;
@@ -23,23 +20,29 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.springframework.integration.splunk.core.Connection;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.integration.splunk.core.ServiceFactory;
import com.splunk.Service;
/**
* Connection to Splunk service
*
* @author Jarred Li
* @since 1.0
* A {@link FactoryBean} for creating a {@link Service}
*
* @author David Turanski
*/
public class SplunkConnection implements Connection<Service> {
public class SplunkServiceFactory implements ServiceFactory {
private final SplunkServer splunkServer;
private Service service;
public SplunkConnection(SplunkServer splunkServer) {
public SplunkServiceFactory(SplunkServer splunkServer) {
this.splunkServer = splunkServer;
}
@Override
public synchronized Service getService() {
if (service != null) {
return service;
}
final Map<String, Object> args = new HashMap<String, Object>();
if (splunkServer.getHost() != null) {
args.put("host", splunkServer.getHost());
@@ -59,15 +62,15 @@ public class SplunkConnection implements Connection<Service> {
args.put("username", splunkServer.getUsername());
args.put("password", splunkServer.getPassword());
ExecutorService executor = Executors.newSingleThreadExecutor();
Future<Service> future = executor.submit(new Callable<Service>(){
public Service call() throws Exception {
return Service.connect(args);
}
});
try {
if (splunkServer.getTimeout() > 0) {
service = future.get(splunkServer.getTimeout(),TimeUnit.MILLISECONDS);
@@ -78,35 +81,6 @@ public class SplunkConnection implements Connection<Service> {
throw new RuntimeException(String.format("could not connect to Splunk Server @ %s:%d - %s",
splunkServer.getHost(),splunkServer.getPort(),e.getMessage()));
}
}
/* (non-Javadoc)
* @see org.springframework.integration.splunk.core.IService#close()
*/
public void close() {
service.logout();
}
/* (non-Javadoc)
* @see org.springframework.integration.splunk.core.IService#isOpen()
*/
public boolean isOpen() {
boolean result = true;
try {
service.getApplications();
} catch (Throwable t) {
result = false;
}
return result;
}
/* (non-Javadoc)
* @see org.springframework.integration.splunk.core.IService#getService()
*/
public Service getTarget() {
return service;
}
}

6
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/support/SplunkSubmitWriter.java
View File

@@ -15,7 +15,7 @@ package org.springframework.integration.splunk.support;
import java.io.IOException;
import java.net.Socket;
import org.springframework.integration.splunk.core.ConnectionFactory;
import org.springframework.integration.splunk.core.ServiceFactory;
import org.springframework.integration.splunk.event.SplunkEvent;
import com.splunk.Args;
@@ -32,8 +32,8 @@ public class SplunkSubmitWriter extends AbstractSplunkDataWriter {
/**
* @param connectionFactory
*/
public SplunkSubmitWriter(ConnectionFactory<Service> connectionFactory, Args args) {
super(connectionFactory, args);
public SplunkSubmitWriter(ServiceFactory serviceFactory, Args args) {
super(serviceFactory, args);
}
/* (non-Javadoc)

6
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/support/SplunkTcpWriter.java
View File

@@ -15,7 +15,7 @@ package org.springframework.integration.splunk.support;
import java.io.IOException;
import java.net.Socket;
import org.springframework.integration.splunk.core.ConnectionFactory;
import org.springframework.integration.splunk.core.ServiceFactory;
import org.springframework.util.Assert;
import com.splunk.Args;
@@ -35,8 +35,8 @@ public class SplunkTcpWriter extends AbstractSplunkDataWriter {
* @param connectionFactory
* @param args
*/
public SplunkTcpWriter(ConnectionFactory<Service> connectionFactory, Args args) {
super(connectionFactory, args);
public SplunkTcpWriter(ServiceFactory serviceFactory, Args args) {
super(serviceFactory, args);
}
/* (non-Javadoc)

8
spring-integration-splunk/src/main/resources/org/springframework/integration/splunk/config/xml/spring-integration-splunk-1.0.xsd
View File

@@ -387,14 +387,6 @@ Indicates the connection timeout in ms.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="pool-server-connection" use="optional" default="true"
type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Whether pool the Splunk connection.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:attributeGroup>
</xsd:schema>