beangle.task.agent source code

1 module beangle.task.agent;
2 
3 import vibe.core.net;
4 import vibe.core.core;
5 import vibe.core.log;
6 import vibe.http.client;
7 import vibe.http.websockets : WebSocket, connectWebSocket;
8 import std.stdio;
9 import std.file;
10 import std.path;
11 import std.string;
12 import std.algorithm;
13 import std.array;
14 import core.time;
15 import beangle.task.config;
16 import beangle.util;
17 
18 int startAgent(string xml, string[] args) {
19   auto agent = Agent.parse(xml);
20   if (agent.serverURL !is null){
21     runTask({
22       try{
23         connectToWS(agent);
24       } catch (Exception e) {
25       }
26     });
27   }
28   if (agent.tasks.length>0){
29     runTask({
30       try{
31         runLocalTask(agent);
32       } catch (Exception e) {
33         assert(false, e.msg);
34       }
35     });
36   }
37   scope (exit) exitEventLoop(true);
38   return runApplication(&args);
39 }
40 
41 void runLocalTask(Agent agent) @system {
42   while(true){
43     sleep(1.seconds);
44     auto current = now();
45     auto interval = 1.seconds;
46     beangle.task.config.Task[] finished;
47     foreach (name, task; agent.tasks){
48       auto next = task.nextExecuteAt();
49       if (next < current){
50         logInfo("finished task: %s", task.name);
51         finished ~= task;
52       }else if ((next-current) <= interval){
53         logInfo("execute task: %s", task.name);
54         task.execute();
55       }
56     }
57     agent.removeTasks(finished);
58   }
59 }
60 
61 void connectToWS (Agent agent)  {
62   if (agent.serverURL is null){
63     logInfo("Ignore remote server connection");
64   }else {
65     while(true){
66       try{
67         logInfo("connect to " ~ agent.serverURL);
68         auto ws = connectWebSocket(URL(agent.serverURL));
69         serveSocket(agent, ws);
70       } catch (Throwable e) {
71         logInfo(e.msg);
72         sleep(10.seconds);
73       }
74     }
75   }
76 }
77 
78 void serveSocket(Agent agent, WebSocket ws){
79   ws.send("/login " ~ agent.name ~ " " ~ agent.secret);
80   while (ws.waitForData()) {
81     auto txt = ws.receiveText().strip();
82     switch (txt){
83       case "tasks" :
84         auto names = agent.tasks.keys.join(",");
85         ws.send(names);
86         break;
87       case "clean" :
88         agent.tasks.clear();
89         ws.send("all tasks were removed.");
90         break;
91         default:
92         if (txt.startsWith("<task")){
93           auto task = parseTask(txt);
94           if (task.repeatable){
95             agent.addTask(task);
96             ws.send("New task " ~ task.name ~ " was mounted.");
97             logInfo("new task %s was mounted.", task.name);
98           }else {
99             auto res = task.execute();
100             string[] outputs;
101             auto f = File(task.stdout, "r");
102             foreach (string l; lines(f)){
103               outputs ~= l;
104             }
105             ws.send(outputs.join());
106             if (res ==0 ){
107               std.file.remove(task.stdout);
108             }
109           }
110         }else if (txt.startsWith("{") && txt.endsWith("}")){
111           import std.json;
112           JSONValue js = parseJSON(txt);
113 
114           auto id = js["id"].str;
115           auto commands = new Commands(js["commands"].str);
116           string[string] envs;
117           auto result = commands.execute(envs, expandTilde("~"));
118           JSONValue jj = ["id":id];
119           jj.object["data"]=result[1];
120           ws.send(jj.toString());
121         }else {
122           logInfo("%s", txt);
123         }
124     }
125   }
126   logFatal("Connection lost!");
127 }