import com.splunk.Service; //导入依赖的package包/类
private List<SplunkEvent> runQuery(JobArgs queryArgs, boolean realtime, SplunkResultProcessor callback) throws Exception {
    Service service = endpoint.getService();
    Job job = service.getJobs().create(getSearch(), queryArgs);
    LOG.debug("Running search : {} with queryArgs : {}", getSearch(), queryArgs);
    if (realtime) {
        while (!job.isReady()) {
            Thread.sleep(500);
        }
        // Besides job.isReady there must be some delay before real time job
        // is ready
        // TODO seems that the realtime stream is not quite isReady to be
        // read
        Thread.sleep(4000);
    } else {
        while (!job.isDone()) {
            Thread.sleep(500);
        }
    }
    return extractData(job, realtime, callback);
}
 

import com.splunk.Service; //导入依赖的package包/类
@Override
protected Socket createSocket(Service service) throws IOException {
    Index indexObject = null;
    Receiver receiver = null;
    Socket socket = null;

    if (index != null) {
        indexObject = service.getIndexes().get(index);
        if (indexObject == null) {
            throw new RuntimeCamelException(String.format("cannot find index [%s]", index));
        }
        socket = indexObject.attach(args);
    } else {
        receiver = service.getReceiver();
        socket = receiver.attach(args);
    }
    socket.setTcpNoDelay(true);
    LOG.trace(String.format("created a socket on %s", socket.getRemoteSocketAddress()));
    return socket;
}
 

import com.splunk.Service; //导入依赖的package包/类
@Test
public void createConsumerEndpointWithMaximalConfiguration() throws Exception {
    SplunkComponent component = context.getComponent("splunk", SplunkComponent.class);

    SplunkEndpoint endpoint = (SplunkEndpoint)component
        .createEndpoint("splunk://normal?username=test&password=pw&host=myhost&port=3333&delay=10s&" + "search=Splunk search query goes here&initEarliestTime=-1d"
                        + "&latestTime=now&count=10&" + "owner=me&app=fantasticapp");
    assertEquals("myhost", endpoint.getConfiguration().getHost());
    assertEquals(3333, endpoint.getConfiguration().getPort());
    assertEquals("test", endpoint.getConfiguration().getUsername());
    assertEquals("pw", endpoint.getConfiguration().getPassword());
    assertEquals("-1d", endpoint.getConfiguration().getInitEarliestTime());
    assertEquals("now", endpoint.getConfiguration().getLatestTime());
    assertEquals(10, endpoint.getConfiguration().getCount());
    assertEquals("me", endpoint.getConfiguration().getOwner());
    assertEquals("fantasticapp", endpoint.getConfiguration().getApp());
    assertEquals(Service.DEFAULT_SCHEME, endpoint.getConfiguration().getScheme());
    assertEquals("Splunk search query goes here", endpoint.getConfiguration().getSearch());
}
 

import com.splunk.Service; //导入依赖的package包/类
/**
 * Create connection with the splunk server.
 */
@Override
public void connect()
{
  try {
    ServiceArgs loginArgs = new ServiceArgs();
    loginArgs.setUsername(userName);
    loginArgs.setPassword(password);
    loginArgs.setHost(host);
    loginArgs.setPort(port);

    service = Service.connect(loginArgs);
  } catch (Exception e) {
    throw new RuntimeException("closing connection", e);
  }
}
 

import com.splunk.Service; //导入依赖的package包/类
/**
 * Returns UP if the information can be obtained and
 * the metrics file can be read.
 */
@Override
public AvailabilityType getAvailability() {
    try {
        log.debug("probe");
        Service service = SplunkDiscovery.service(config);
        serviceInfo = service.getInfo();
        service.logout();
        /*
        Splunk file rotation makes the metrics file unreadable
        This is Splunk case 117457
        log.debug("tail");
        tail.read();
        */
        log.debug("done");
        return AvailabilityType.UP;
    } catch (Exception e) {
        log.debug("failed checking splunk", e);
        return AvailabilityType.DOWN;
    }
}
 

import com.splunk.Service; //导入依赖的package包/类
public int getNumberOfResultsFromJob(SplunkConf conf) {
    Service service = job.getService();
    String searchString = job.getSearch();
    searchString = searchString.concat(" | stats count");
    SplunkConf newConfig = new SplunkConf(conf);
    newConfig.set(SplunkConf.SPLUNK_SEARCH_QUERY,searchString);

    SplunkJob getEvents = SplunkJob.createSplunkJob(service,newConfig);
    getEvents.waitForCompletion(100);

    JobResultsArgs resultsArgs = new JobResultsArgs();
    resultsArgs.setOutputMode(JobResultsArgs.OutputMode.CSV);
    resultsArgs.setFieldList(new String[] {"count"});

    BufferedReader br = new BufferedReader(new InputStreamReader(getEvents.getJob().getResults(resultsArgs)));
    try {
        br.readLine(); // Skip header
        return Integer.parseInt(br.readLine());
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}
 

import com.splunk.Service; //导入依赖的package包/类
@Override
protected Socket createSocket(Service service) throws IOException {
    TcpInput input = (TcpInput)service.getInputs().get(String.valueOf(port));
    if (input == null) {
        throw new RuntimeException("no input defined for port " + port);
    }
    if (input.isDisabled()) {
        throw new RuntimeException(String.format("input on port %d is disabled", port));
    }
    return input.attach();
}
 

import com.splunk.Service; //导入依赖的package包/类
@Test
public void createProducerEndpointWithMinimalConfiguration() throws Exception {
    SplunkComponent component = context.getComponent("splunk", SplunkComponent.class);

    SplunkEndpoint endpoint = (SplunkEndpoint)component.createEndpoint("splunk://test?username=test&password=pw");
    assertEquals(Service.DEFAULT_HOST, endpoint.getConfiguration().getHost());
    assertEquals(Service.DEFAULT_PORT, endpoint.getConfiguration().getPort());
    assertEquals("test", endpoint.getConfiguration().getUsername());
    assertEquals("pw", endpoint.getConfiguration().getPassword());
    assertEquals(Service.DEFAULT_SCHEME, endpoint.getConfiguration().getScheme());
    assertEquals(5000, endpoint.getConfiguration().getConnectionTimeout());
    assertFalse(endpoint.getConfiguration().isUseSunHttpsHandler());
    assertFalse(endpoint.getConfiguration().isRaw());
    assertEquals(SSLSecurityProtocol.TLSv1_2, endpoint.getConfiguration().getSslProtocol());
}
 

import com.splunk.Service; //导入依赖的package包/类
@Test
public void createConsumerEndpointWithMinimalConfiguration() throws Exception {
    SplunkComponent component = context.getComponent("splunk", SplunkComponent.class);

    SplunkEndpoint endpoint = (SplunkEndpoint)component.createEndpoint("splunk://realtime?username=test&" + "password=pw&search=Splunk search query goes here");
    assertEquals(Service.DEFAULT_HOST, endpoint.getConfiguration().getHost());
    assertEquals(Service.DEFAULT_PORT, endpoint.getConfiguration().getPort());
    assertEquals("test", endpoint.getConfiguration().getUsername());
    assertEquals("pw", endpoint.getConfiguration().getPassword());
    assertEquals(Service.DEFAULT_SCHEME, endpoint.getConfiguration().getScheme());
    assertEquals(5000, endpoint.getConfiguration().getConnectionTimeout());
    assertEquals("Splunk search query goes here", endpoint.getConfiguration().getSearch());
}
 

import com.splunk.Service; //导入依赖的package包/类
public static DefaultSplunkConfigurationFactory mockComponent(final Service service, final Socket socket) {
    return new DefaultSplunkConfigurationFactory() {
        @Override
        public SplunkConfiguration parseMap(Map<String, Object> parameters) {
            return new MockConnectionSettings(service, socket);
        }
    };
}
 

import com.splunk.Service; //导入依赖的package包/类
public SplunkReporter(Builder builder, Service splunk) {
    super(builder.registry, builder.prefix, builder.filter, builder.rateUnit, builder.durationUnit);
    receiver = splunk.getReceiver();
    this.showSamples = builder.showSamples;
    splunkArgs.put("source", builder.source);
    splunkArgs.put("sourcetype", builder.sourcetype);
    this.index = builder.index;
    this.extraAttributes = builder.extraAttributes;
    mapper = new ObjectMapper().registerModule(new MetricsModule(builder.rateUnit, builder.durationUnit, this.showSamples));
}
 

import com.splunk.Service; //导入依赖的package包/类
public Service getService() {
    if (service == null) {
        this.service = configuration.getConnectionFactory().createService(getCamelContext());
    }
    return service;
}
 

import com.splunk.Service; //导入依赖的package包/类
MockConnectionSettings(Service service, Socket socket) {
    this.service = service;
    this.socket = socket;
    mockSplunkWriterApi();
    setConnectionFactory(new MockConnectionFactory(service));
}
 

import com.splunk.Service; //导入依赖的package包/类
MockConnectionFactory(Service service) {
    super("foo", "bar");
    this.service = service;
}
 

import com.splunk.Service; //导入依赖的package包/类
@Override
public Service createService(CamelContext camelContext) {
    return service;
}
 

import com.splunk.Service; //导入依赖的package包/类
public Service getService()
{
  return service;
}
 

import com.splunk.Service; //导入依赖的package包/类
@Before
public void setup() throws Exception {
    splunk = mock(Service.class);
    receiver = mock(Receiver.class);
    when(splunk.getReceiver()).thenReturn(receiver);

    reg = new MetricRegistry();

    reg.counter("tcounter").inc();
    reg.meter("tmeter").mark();
    reg.histogram("thist").update(10);
    reg.histogram("thist").update(1);
    reg.timer("ttimer").time().stop();
}
 

import com.splunk.Service; //导入依赖的package包/类
public static void main(String[] args) {
	// Create login parameters. We suggest finding
	// a better way to store these than hard coding
	// them in your program for production code.
	ServiceArgs serviceArgs = new ServiceArgs();
	serviceArgs.setUsername("admin");
	serviceArgs.setPassword("changeme");
	serviceArgs.setHost("localhost");
	serviceArgs.setPort(8089);

	// Create a Service instance and log in with the argument map
	Service service = Service.connect(serviceArgs);
	
	// Set the parameters for the search
	Args oneshotSearchArgs = new Args();
	// For a full list of options, see:
	//
	//     http://docs.splunk.com/Documentation/Splunk/latest/RESTAPI/RESTsearch#POST_search.2Fjobs

	// oneshotSearchArgs.put("earliest_time", "now-1w");
	// oneshotSearchArgs.put("latest_time",   "now");

	InputStream resultsStream = service.oneshotSearch("search * | head 5",
			oneshotSearchArgs);

	try {
		ResultsReaderXml resultsReader = new ResultsReaderXml(resultsStream);

		for (Event event : resultsReader) {
			// Process each event
			for (String key: event.keySet()) {
			   System.out.println(key + ": " + event.get(key));
			}
			System.out.println("");
		}

		resultsReader.close();
	} catch (IOException e) {
		e.printStackTrace();
	}
}
 

import com.splunk.Service; //导入依赖的package包/类
public static SplunkJob getSplunkJob(Service service, String jobId) {
    Job job = service.getJobs().get(jobId);
    return new SplunkJob(job);
}
 

import com.splunk.Service; //导入依赖的package包/类
public static Service connect(SplunkConf configuration) {
    return connect(
            configuration.get(SplunkConf.SPLUNK_USERNAME),
            configuration.get(SplunkConf.SPLUNK_PASSWORD),
            configuration.get(SplunkConf.SPLUNK_HOST),
            configuration.getInt(SplunkConf.SPLUNK_PORT, 8080));
}