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

Merge pull request #26 from dturanski/INTEXT-43

Browse Source 
* dturanski-INTEXT-43:
  INTEXT-43 replaced ConnectionFactory with ServiceFactory
  INTEXT-42 refactored SplunkEvent
  Updated README
  Updated README
This commit is contained in:
Gunnar Hillert
2013-02-08 11:51:47 -05:00
parent 978d80a9fb 885e2ebc11
commit f2d3a87946
32 changed files with 315 additions and 858 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
spring-integration-splunk/README.md
View File

@@ -39,7 +39,7 @@ The Inbound channel adapter is used to read data from Splunk and output a messag
### Saved search:
```xml
<int-splunk:inbound-channel-adapter id="splunkInboundChannelAdapter"
savedSearch="test" splunk-server-ref="splunkServer"
@@ -61,7 +61,7 @@ The Inbound channel adapter is used to read data from Splunk and output a messag
### Export:
```xml
```xml
<int-splunk:inbound-channel-adapter id="splunkInboundChannelAdapter"
auto-startup="true" search="search spring:example" splunk-server-ref="splunkServer" channel="inputFromSplunk"
mode="EXPORT" earliestTime="-5d" latestTime="now" initEarliestTime="-1d">
@@ -86,8 +86,8 @@ The outbound channel adapter requires a child *-writer element which defines rel
<int-splunk:outbound-channel-adapter
id="splunkOutboundChannelAdapter"
channel="outputToSplunk"
splunk-server-ref="splunkServer"
sourceType="spring-integration"
splunk-server-ref="splunkServer"
sourceType="spring-integration"
source="example2">
<int-splunk:submit-writer index="foo"/>
</int-splunk:outbound-channel-adapter>
@@ -98,7 +98,7 @@ The outbound channel adapter requires a child *-writer element which defines rel
```xml
<int-splunk:outbound-channel-adapter
id="splunkOutboundChannelAdapter"
channel="outputToSplunk"
channel="outputToSplunk"
splunk-server-ref="splunkServer"
>
<int-splunk:index-writer index="someIndex"/>
@@ -109,30 +109,30 @@ The outbound channel adapter requires a child *-writer element which defines rel
```xml
<int-splunk:outbound-channel-adapter
id="splunkOutboundChannelAdapter"
channel="outputToSplunk"
id="splunkOutboundChannelAdapter"
channel="outputToSplunk"
splunk-server-ref="splunkServer"
>
<int-splunk:tcp-writer port="9999"/>
</int-splunk:outbound-channel-adapter>
```
*NOTE: The input must exist and be enabled on the server*
### Configuring The Splunk Server connection
```xml
<int-splunk:server id="splunkServer" username="admin" password="password" timeout="5000" host="somehost.someplace.com" port="9000" />
```
Additional server properties include (see [splunk](http://docs.splunk.com/Documentation/Splunk/latest) documentation for details):
```
Additional server properties include (see [splunk](http://docs.splunk.com/Documentation/Splunk/latest) documentation for details):
* app
* scheme
* scope
* owner
The default host is *localhost* and the default port is *8089*. The *timeout* attribute indicates how long to wait for a connection in miliseconds.
The default host is *localhost* and the default port is *8089*. The *timeout* attribute indicates how long to wait for a connection in miliseconds.
Development
@@ -146,3 +146,4 @@ Development
To generate Eclipse metadata (e.g., .classpath and .project files), do the following:
./gradlew eclipse

2
spring-integration-splunk/build.gradle
View File

@@ -36,7 +36,7 @@ dependencies {
compile "org.springframework:spring-context:$springVersion"
compile "org.springframework:spring-expression:$springVersion"
compile "org.springframework.integration:spring-integration-core:$springIntegrationVersion"
compile "commons-lang:commons-lang:$commonsLangVersion"
compile "joda-time:joda-time:$jodaTimeVersion"
compile "commons-pool:commons-pool:$commonsPoolVersion"
testCompile "org.mockito:mockito-all:$mockitoVersion"

8
spring-integration-splunk/gradle.properties
View File

@@ -1,10 +1,10 @@
version=0.5.0.BUILD-SNAPSHOT
version=1.0.0.BUILD-SNAPSHOT
cglibVersion=2.2
junitVersion=4.8.2
log4jVersion=1.2.12
mockitoVersion=1.9.0
springVersion=3.1.2.RELEASE
springIntegrationVersion=2.1.2.RELEASE
commonsLangVersion=2.6
springVersion=3.1.3.RELEASE
springIntegrationVersion=2.2.1.RELEASE
jodaTimeVersion=2.1
commonsPoolVersion=1.6
splunkVersion =1.0.0

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();
}

99
spring-integration-splunk/src/main/java/org/springframework/integration/splunk/event/SplunkEvent.java
View File

@@ -17,15 +17,19 @@ package org.springframework.integration.splunk.event;
import java.io.Serializable;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.lang.time.FastDateFormat;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
import org.springframework.util.Assert;
/**
* Splunk data entity
*
* @author Jarred Li
* @author Damien Dallimore damien@dtdsoftware.com
* @author David Turanski
* @since 1.0
*
*/
@@ -33,8 +37,6 @@ import org.apache.commons.lang.time.FastDateFormat;
@SuppressWarnings("serial")
public class SplunkEvent implements Serializable {
private Map<String, String> eventData;
/**
* Contents of the event message
*/
@@ -65,11 +67,11 @@ public class SplunkEvent implements Serializable {
/**
* default date format is using internal generated date
*/
protected static final String DATEFORMATPATTERN = "yyyy-MM-dd HH:mm:ss:SSSZ";
protected static final String DATEFORMATPATTERN = "yyyy-MM-dd\tHH:mm:ss:SSSZ";
/**
* Date Formatter instance
* Date Formatter
*/
protected static FastDateFormat DATEFORMATTER = FastDateFormat.getInstance(DATEFORMATPATTERN);
protected static final DateTimeFormatter DATE_FORMATTER = DateTimeFormat.forPattern(DATEFORMATPATTERN);
/**
* Event prefix fields
@@ -85,7 +87,7 @@ public class SplunkEvent implements Serializable {
protected static final String THROWABLE_STACKTRACE_ELEMENTS = "stacktrace_elements";
protected static final String LINEBREAK = "\n";
// ----------------------------------
// Common event fields
// ----------------------------------
@@ -227,26 +229,33 @@ public class SplunkEvent implements Serializable {
*/
public static String UPDATE_PACKAGE = "package";
/**
* A Constructor to load data from a Map
* @param data the map
*/
public SplunkEvent(Map<String, String> data) {
this.eventMessage = new StringBuffer();
this.eventData = data;
for (String key : data.keySet()) {
this.addPair(key, data.get(key));
}
}
/**
* Constructor.
*
* @param eventName
* the event name
* @param eventID
* the event id
* @param useInternalDate
* Whether or not to add a date to the event string
* @param quoteValues
* Whether or not to put quotes around values
* A Copy constructor
* @param splunkEvent
*/
public SplunkEvent(SplunkEvent splunkEvent) {
this.eventMessage = splunkEvent.eventMessage;
this.quoteValues = splunkEvent.quoteValues;
this.useInternalDate = splunkEvent.useInternalDate;
}
/**
* Constructor to create a generic event
* @param eventName the event name
* @param eventID the event id
* @param useInternalDate whether or not to add a date to the event string
* @param quoteValues whether or not to put quotes around values
*/
public SplunkEvent(String eventName, String eventID, boolean useInternalDate, boolean quoteValues) {
@@ -259,13 +268,10 @@ public class SplunkEvent implements Serializable {
}
/**
* Constructor.Will add internally generated date and put quotes around
* values.
* Constructor to create a generic event with the default format
*
* @param eventName
* the event name
* @param eventID
* the event ID
* @param eventName the event name
* @param eventID the event ID
*/
public SplunkEvent(String eventName, String eventID) {
@@ -279,20 +285,27 @@ public class SplunkEvent implements Serializable {
this.eventMessage = new StringBuffer();
}
/**
* Simple shallow cloning method
*/
public SplunkEvent clone() {
SplunkEvent clone = new SplunkEvent();
clone.quoteValues = this.quoteValues;
clone.useInternalDate = this.useInternalDate;
clone.eventMessage.append(this.eventMessage);
return clone;
}
public Map<String, String> getEventData() {
Map<String, String> eventData = new HashMap<String, String>();
String eventEntries = eventMessage.toString();
String[] entries = eventEntries.split(PAIRDELIM);
String quote = new String(new char[] { QUOTE });
for (String entry : entries) {
String[] pair = entry.split(KVDELIM);
Assert.isTrue(pair.length == 2, String.format("invalid event data [%s]", entry));
String key = pair[0].replaceAll(quote, "");
String value = pair[1].replaceAll(quote, "");
if ("null".equals(value)) {
value = null;
}
eventData.put(key, value);
}
return eventData;
}
@@ -417,7 +430,7 @@ public class SplunkEvent implements Serializable {
* @param value
*/
public void addPair(String key, String value) {
Assert.notNull(key, "key cannot be null");
if (quoteValues)
this.eventMessage.append(key).append(KVDELIM).append(QUOTE).append(value).append(QUOTE).append(PAIRDELIM);
else
@@ -435,13 +448,13 @@ public class SplunkEvent implements Serializable {
if (useInternalDate) {
StringBuffer clonedMessage = new StringBuffer();
clonedMessage.append(DATEFORMATTER.format(new Date())).append(PAIRDELIM).append(this.eventMessage);
clonedMessage.append(DATE_FORMATTER.print(new Date().getTime())).append(PAIRDELIM)
.append(this.eventMessage);
event = clonedMessage.toString();
}
else
} else
event = eventMessage.toString();
// trim off trailing pair delim char(s)
String result = event.substring(0, event.length() - PAIRDELIM.length()) + LINEBREAK;
String result = event.substring(0, event.length() - PAIRDELIM.length()) + LINEBREAK;
return result;
}

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>

4
spring-integration-splunk/src/test/java/org/springframework/integration/splunk/config/xml/SplunkInboundChannelAdapterParserSavedSearchTests.java
View File

@@ -15,8 +15,8 @@
*/
package org.springframework.integration.splunk.config.xml;
import static org.junit.Assert.*;
import junit.framework.Assert;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import org.junit.Test;
import org.junit.runner.RunWith;

67
spring-integration-splunk/src/test/java/org/springframework/integration/splunk/event/SplunkEventTests.java Normal file
View File

@@ -0,0 +1,67 @@
/*
* 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.event;
import static org.junit.Assert.assertEquals;
import java.util.HashMap;
import java.util.Map;
import org.junit.Test;
/**
* @author David Turanski
*
*/
public class SplunkEventTests {
@Test
public void testGetEventData() {
Map<String,String> data = new HashMap<String,String>();
data.put("foo", "foo");
data.put("bar", null);
SplunkEvent event = new SplunkEvent(data);
Map<String,String> eventData = event.getEventData();
assertEquals(data.size(),eventData.size());
for (String key: eventData.keySet()) {
assertEquals(data.get(key),eventData.get(key));
}
}
public void testGetEventDataEmpty() {
SplunkEvent event = new SplunkEvent( );
Map<String,String> eventData = event.getEventData();
assertEquals(0,eventData.size());
}
@Test(expected=RuntimeException.class)
public void testKeyCannotBeNull() {
SplunkEvent event = new SplunkEvent();
event.addPair(null, "foo");
}
@Test
public void testCopyConstructor() {
Map<String,String> data = new HashMap<String,String>();
data.put("foo", "foo");
data.put("bar", "bar");
SplunkEvent event = new SplunkEvent(data);
SplunkEvent event2 = new SplunkEvent(event);
assertEquals(event.quoteValues, event2.quoteValues);
assertEquals(event.useInternalDate,event2.useInternalDate);
Map<String,String> eventData = event.getEventData();
Map<String,String> event2Data = event2.getEventData();
assertEquals(eventData.size(),event2Data.size());
for (String key: eventData.keySet()) {
assertEquals(eventData.get(key),event2Data.get(key));
}
}
}

2
spring-integration-splunk/src/test/java/org/springframework/integration/splunk/outbound/SplunkOutboundChannelAdapterTests.java
View File

@@ -47,7 +47,7 @@ public class SplunkOutboundChannelAdapterTests {
@Test
public void testHandleRequestMessage() {
Message<?> message = null;
when(executor.executeOutboundOperation(message)).thenReturn(null);
when(executor.write(message)).thenReturn(null);
Object ret = outboundAdapter.handleRequestMessage(message);
Assert.assertNull(ret);

87
spring-integration-splunk/src/test/java/org/springframework/integration/splunk/support/ConnectionFactoryFactoryBeanTests.java
View File

@@ -1,87 +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 junit.framework.Assert;
import org.junit.Test;
import org.springframework.integration.splunk.core.ConnectionFactory;
import com.splunk.Service;
/**
* @author Jarred Li
* @since 1.0
*
*/
public class ConnectionFactoryFactoryBeanTests {
private ConnectionFactoryFactoryBean<Service> factoryBean;
/**
* Test method for {@link org.springframework.integration.splunk.support.ConnectionFactoryFactoryBean#ConnectionFactoryFactoryBean(org.springframework.integration.splunk.core.ConnectionFactory, boolean)}.
* @throws Exception
*/
@Test
public void testConnectionFactoryFactoryBean() throws Exception {
SplunkServer server = new SplunkServer();
SplunkConnectionFactory factory = new SplunkConnectionFactory(server);
factoryBean = new ConnectionFactoryFactoryBean<Service>(factory, false);
ConnectionFactory<Service> generatedByFactoryBean = factoryBean.getObject();
Assert.assertTrue(generatedByFactoryBean instanceof SplunkConnectionFactory);
}
@Test
public void testConnectionFactoryFactoryBean_withPoll() throws Exception {
SplunkServer server = new SplunkServer();
SplunkConnectionFactory factory = new SplunkConnectionFactory(server);
factoryBean = new ConnectionFactoryFactoryBean<Service>(factory, true);
ConnectionFactory<Service> generatedByFactoryBean = factoryBean.getObject();
Assert.assertTrue(generatedByFactoryBean instanceof PoolingConnectionFactory);
}
/**
* Test method for {@link org.springframework.integration.splunk.support.ConnectionFactoryFactoryBean#getObjectType()}.
*/
@Test
public void testGetObjectType() {
SplunkServer server = new SplunkServer();
SplunkConnectionFactory factory = new SplunkConnectionFactory(server);
factoryBean = new ConnectionFactoryFactoryBean<Service>(factory, true);
Class<?> clazz = factoryBean.getObjectType();
Assert.assertEquals(PoolingConnectionFactory.class, clazz);
}
/**
* Test method for {@link org.springframework.integration.splunk.support.ConnectionFactoryFactoryBean#isSingleton()}.
*/
@Test
public void testIsSingleton() {
SplunkServer server = new SplunkServer();
SplunkConnectionFactory factory = new SplunkConnectionFactory(server);
factoryBean = new ConnectionFactoryFactoryBean<Service>(factory, false);
Assert.assertTrue(factoryBean.isSingleton());
}
}

111
spring-integration-splunk/src/test/java/org/springframework/integration/splunk/support/PoolingConnectionFactoryTests.java
View File

@@ -1,111 +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 static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import junit.framework.Assert;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.splunk.core.Connection;
import org.springframework.integration.splunk.core.ConnectionFactory;
/**
* @author Jarred Li
* @since 1.0
*
*/
public class PoolingConnectionFactoryTests {
private ConnectionFactory<TestEntity> conFactory;
private PoolingConnectionFactory<TestEntity> poolConFactory;
@SuppressWarnings("unchecked")
@Before
public void before() {
conFactory = mock(ConnectionFactory.class);
poolConFactory = new PoolingConnectionFactory<TestEntity>(conFactory);
}
/**
* Test method for {@link org.springframework.integration.splunk.support.PoolingConnectionFactory#getConnection()}.
* @throws Exception
*/
@Test
public void testGetConnection() throws Exception {
@SuppressWarnings("unchecked")
Connection<TestEntity> con = mock(Connection.class);
when(con.getTarget()).thenReturn(new TestEntity("entity1"));
when(conFactory.getConnection()).thenReturn(con);
Connection<TestEntity> returnCon = poolConFactory.getConnection();
TestEntity obj1 = returnCon.getTarget();
Assert.assertNotNull(obj1);
Assert.assertEquals("entity1", obj1.getName());
when(con.getTarget()).thenReturn(new TestEntity("entity2"));
when(conFactory.getConnection()).thenReturn(con);
returnCon = poolConFactory.getConnection();
TestEntity obj2 = returnCon.getTarget();
Assert.assertNotNull(obj2);
Assert.assertEquals("entity2", obj2.getName());
when(con.getTarget()).thenReturn(new TestEntity("entity3"));
when(conFactory.getConnection()).thenReturn(con);
returnCon = poolConFactory.getConnection();
TestEntity obj3 = returnCon.getTarget();
Assert.assertNotNull(obj3);
Assert.assertEquals("entity3", obj3.getName());
}
/**
* Test method for {@link org.springframework.integration.splunk.support.PoolingConnectionFactory#destroy()}.
* @throws Exception
*/
@Test(expected = IllegalStateException.class)
public void testDestroy() throws Exception {
try {
poolConFactory.destroy();
} catch (Exception e) {
fail("exception when destoying pool connection factory");
}
poolConFactory.getConnection();
}
public static class TestEntity {
private String name;
public TestEntity(String n) {
this.name = n;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
}
}

92
spring-integration-splunk/src/test/java/org/springframework/integration/splunk/support/SplunkDataReaderTests.java
View File

@@ -15,10 +15,6 @@
*/
package org.springframework.integration.splunk.support;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
@@ -29,9 +25,9 @@ import junit.framework.Assert;
import org.junit.Before;
import org.junit.Test;
import static org.mockito.Mockito.*;
import org.springframework.core.io.ClassPathResource;
import org.springframework.integration.splunk.core.Connection;
import org.springframework.integration.splunk.core.ConnectionFactory;
import org.springframework.integration.splunk.core.ServiceFactory;
import org.springframework.integration.splunk.event.SplunkEvent;
import com.splunk.Job;
@@ -49,7 +45,8 @@ public class SplunkDataReaderTests {
@Before
public void before() {
reader = new SplunkDataReader(new TestConnectioniFactory());
ServiceFactory serviceFactory =serviceFactory();
reader = new SplunkDataReader(serviceFactory);
}
/**
@@ -60,7 +57,7 @@ public class SplunkDataReaderTests {
public void testBlockingSearch() throws Exception {
reader.setMode(SearchMode.BLOCKING);
reader.setSearch("search spring:example");
List<SplunkEvent> data = reader.search();
List<SplunkEvent> data = reader.read();
Assert.assertNotNull(data);
Assert.assertEquals(5, data.size());
}
@@ -69,7 +66,7 @@ public class SplunkDataReaderTests {
public void testNonBlockingSearch() throws Exception {
reader.setMode(SearchMode.NORMAL);
reader.setSearch("search spring:example");
List<SplunkEvent> data = reader.search();
List<SplunkEvent> data = reader.read();
Assert.assertNotNull(data);
Assert.assertEquals(5, data.size());
}
@@ -79,65 +76,34 @@ public class SplunkDataReaderTests {
public void testRealtimeSearch() throws Exception {
reader.setMode(SearchMode.REALTIME);
reader.setSearch("search spring:example");
List<SplunkEvent> data = reader.search();
List<SplunkEvent> data = reader.read();
Assert.assertNotNull(data);
Assert.assertEquals(5, data.size());
}
public static class TestConnectioniFactory implements ConnectionFactory<Service> {
private ServiceFactory serviceFactory() {
InputStream is = null;
/* (non-Javadoc)
* @see org.springframework.integration.splunk.core.ConnectionFactory#getConnection()
*/
public Connection<Service> getConnection() throws Exception {
return new TestConnection();
try {
is = new ClassPathResource("splunk-data.xml").getInputStream();
} catch (FileNotFoundException e) {
Assert.fail("can not read splunk data file");
} catch (IOException e) {
Assert.fail("can not read splunk data file");
}
Service service = mock(Service.class);
service.setToken("token");
JobCollection jobCollection = mock(JobCollection.class);
Job blockingJob = mock(Job.class);
when(blockingJob.isDone()).thenReturn(true);
when(blockingJob.getResultCount()).thenReturn(5);
when(blockingJob.getResults(any(Map.class))).thenReturn(is);
when(jobCollection.create(any(String.class), any(Map.class))).thenReturn(blockingJob);
when(service.getJobs()).thenReturn(jobCollection);
ServiceFactory serviceFactory = mock(ServiceFactory.class);
when(serviceFactory.getService()).thenReturn(service);
return serviceFactory;
}
public static class TestConnection implements Connection<Service> {
private Service service = mock(Service.class);
/* (non-Javadoc)
* @see org.springframework.integration.splunk.core.Connection#getTarget()
*/
public Service getTarget() {
InputStream is = null;
try {
is = new ClassPathResource("splunk-data.xml").getInputStream();
} catch (FileNotFoundException e) {
Assert.fail("can not read splunk data file");
} catch (IOException e) {
Assert.fail("can not read splunk data file");
}
service.setToken("token");
JobCollection jobCollection = mock(JobCollection.class);
Job blockingJob = mock(Job.class);
when(blockingJob.isDone()).thenReturn(true);
when(blockingJob.getResultCount()).thenReturn(5);
when(blockingJob.getResults(any(Map.class))).thenReturn(is);
when(jobCollection.create(any(String.class), any(Map.class))).thenReturn(blockingJob);
when(service.getJobs()).thenReturn(jobCollection);
return service;
}
/* (non-Javadoc)
* @see org.springframework.integration.splunk.core.Connection#close()
*/
public void close() {
}
/* (non-Javadoc)
* @see org.springframework.integration.splunk.core.Connection#isOpen()
*/
public boolean isOpen() {
return true;
}
}
}

62
spring-integration-splunk/src/test/java/org/springframework/integration/splunk/support/SplunkDataWriterTests.java
View File

@@ -15,13 +15,15 @@
*/
package org.springframework.integration.splunk.support;
import static org.mockito.Matchers.*;
import static org.mockito.Mockito.*;
import static org.mockito.Matchers.eq;
import static org.mockito.Matchers.matches;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.splunk.core.Connection;
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;
@@ -40,10 +42,11 @@ public class SplunkDataWriterTests {
private static Receiver receiver = mock(Receiver.class);
private Args args;
@Before
public void before() {
public void before() throws Exception {
args = new Args();
writer = new SplunkSubmitWriter(new TestConnectionFactory(),args);
writer = new SplunkSubmitWriter(serviceFactory(), args);
writer.start();
}
@@ -53,53 +56,24 @@ public class SplunkDataWriterTests {
*/
@Test
public void testWrite() throws Exception {
SplunkEvent sd = new SplunkEvent("spring", "spring:example");
sd.setCommonDesc("description");
writer.write(sd);
verify(receiver).submit(eq(args), matches(".*spring:example.*\n"));
writer.stop();
}
public static class TestConnectionFactory implements ConnectionFactory<Service> {
private ServiceFactory serviceFactory() throws Exception {
ServiceFactory serviceFactory = mock(ServiceFactory.class);
/* (non-Javadoc)
* @see org.springframework.integration.splunk.core.ConnectionFactory#getConnection()
*/
public Connection<Service> getConnection() throws Exception {
return new TestConnection();
}
Service service = mock(Service.class);
service.setToken("token");
when(service.getReceiver()).thenReturn(receiver);
when(serviceFactory.getService()).thenReturn(service);
return serviceFactory;
}
public static class TestConnection implements Connection<Service> {
private Service service = mock(Service.class);
/* (non-Javadoc)
* @see org.springframework.integration.splunk.core.Connection#getTarget()
*/
public Service getTarget() {
service.setToken("token");
when(service.getReceiver()).thenReturn(receiver);
return service;
}
/* (non-Javadoc)
* @see org.springframework.integration.splunk.core.Connection#close()
*/
public void close() {
}
/* (non-Javadoc)
* @see org.springframework.integration.splunk.core.Connection#isOpen()
*/
public boolean isOpen() {
return true;
}
}
}

2
spring-integration-splunk/src/test/java/org/springframework/integration/splunk/support/SplunkExecutorTests.java
View File

@@ -80,7 +80,7 @@ public class SplunkExecutorTests {
sd = new SplunkEvent("spring", "spring:example");
sd.setCommonDesc("description");
data.add(sd);
when(reader.search()).thenReturn(data);
when(reader.read()).thenReturn(data);
List<SplunkEvent> result = executor.poll();
Assert.assertEquals(2, result.size());

2
spring-integration-splunk/src/test/resources/org/springframework/integration/splunk/config/xml/SplunkOutboundChannelAdapterParserStreamTests-context.xml
View File

@@ -16,7 +16,7 @@
<int-splunk:outbound-channel-adapter
id="splunkOutboundChannelAdapter" auto-startup="false" order="1"
channel="outputToSplunk" splunk-server-ref="splunkServer"
pool-server-connection="true" source-type="spring-integration" source="example5"
source-type="spring-integration" source="example5"
host="test.host" host-regex="test.host.*">
<int-splunk:index-writer/>
</int-splunk:outbound-channel-adapter>

2
spring-integration-splunk/src/test/resources/org/springframework/integration/splunk/config/xml/SplunkOutboundChannelAdapterParserTests-context.xml
View File

@@ -16,7 +16,7 @@
<int-splunk:outbound-channel-adapter
id="splunkOutboundChannelAdapter" auto-startup="false" order="1"
channel="outputToSplunk" splunk-server-ref="splunkServer"
pool-server-connection="true" source-type="spring-integration" source="example5"
source-type="spring-integration" source="example5"
>
<int-splunk:submit-writer/>
</int-splunk:outbound-channel-adapter>