summaryrefslogtreecommitdiffstats
path: root/gearman/controllerWorker/ControllerWorker/SomePingWorker.java
blob: f3f6fe6f5c494e6072f3bb9a2f55ab522e512cfe (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
package ControllerWorker;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.ListIterator;
import java.util.concurrent.ExecutionException;

import org.gearman.client.GearmanClient;
import org.gearman.client.GearmanClientImpl;
import org.gearman.client.GearmanJob;
import org.gearman.client.GearmanJobImpl;
import org.gearman.client.GearmanJobResult;
import org.gearman.client.GearmanJobResultImpl;
import org.gearman.client.GearmanJobStatus;
import org.gearman.common.GearmanJobServerConnection;
import org.gearman.common.GearmanNIOJobServerConnection;
import org.gearman.util.ByteUtils;
import org.gearman.worker.AbstractGearmanFunction;

import org.json.simple.JSONArray;
import org.json.simple.JSONValue;

public class SomePingWorker extends AbstractGearmanFunction {
	@Override
	public String getName() {
		return "somePing";
	}

	@Override
	public GearmanJobResult executeFunction() {
		final GearmanJobServerConnection connection = new GearmanNIOJobServerConnection(
				"127.0.0.1", 4730);
		GearmanClient client = new GearmanClientImpl();
		client.addJobServer(connection);
		String data = ByteUtils.fromUTF8Bytes((byte[]) this.data);
		JSONArray jsonArray = (JSONArray) JSONValue.parse(data);
		String res = "";
		List<GearmanJob> jobs = new ArrayList<GearmanJob>();
		int count = 1;

		for (Object ipObj : jsonArray) {
			String ip = ipObj.toString();
			System.out.println("Ping " + ip);
			GearmanJob job = GearmanJobImpl.createJob("ping", ip.getBytes(),
					"ping" + count);
			client.submit(job);
			jobs.add(job);
			count++;
		}

		Boolean exit = false;
		List<GearmanJob> remove = new ArrayList<GearmanJob>();

		while (!exit) {
			ListIterator<GearmanJob> it = jobs.listIterator();

			while (it.hasNext()) {
				GearmanJob j = it.next();
				GearmanJobStatus jStatus;
				try {
					jStatus = client.getJobStatus(j);
					if (!jStatus.isKnown() && j.isDone()) {
						GearmanJobResult jobRes = j.get();
						String jobID = ByteUtils.fromUTF8Bytes(j.getID());
						if (jobID.equals("ping1")) {
							res += ByteUtils.fromUTF8Bytes(jobRes.getResults());
						} else {
							res += "; "
									+ ByteUtils.fromUTF8Bytes(jobRes
											.getResults());
						}
						remove.add(j);
					}
				} catch (IllegalStateException e1) {
					e1.printStackTrace();
				} catch (IOException e1) {
					e1.printStackTrace();
				} catch (InterruptedException e) {
					e.printStackTrace();
				} catch (ExecutionException e) {
					e.printStackTrace();
				}
			}

			ListIterator<GearmanJob> it2 = remove.listIterator();

			while (it2.hasNext()) {
				jobs.remove(it2.next());
			}

			remove.clear();

			if (jobs.isEmpty()) {
				exit = true;
			}
		}

		byte[] warnings = new byte[0];
		byte[] exceptions = new byte[0];
		int numerator = 0;
		int denominator = 0;
		GearmanJobResult gjr = new GearmanJobResultImpl(this.jobHandle, true,
				res.getBytes(), warnings, exceptions, numerator, denominator);
		return gjr;
	}
}