func_provider.go 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927
  1. /*
  2. * ECAL
  3. *
  4. * Copyright 2020 Matthias Ladkau. All rights reserved.
  5. *
  6. * This Source Code Form is subject to the terms of the MIT
  7. * License, If a copy of the MIT License was not distributed with this
  8. * file, You can obtain one at https://opensource.org/licenses/MIT.
  9. */
  10. package interpreter
  11. import (
  12. "fmt"
  13. "strconv"
  14. "strings"
  15. "time"
  16. "devt.de/krotik/common/errorutil"
  17. "devt.de/krotik/common/timeutil"
  18. "devt.de/krotik/ecal/engine"
  19. "devt.de/krotik/ecal/parser"
  20. "devt.de/krotik/ecal/scope"
  21. "devt.de/krotik/ecal/stdlib"
  22. "devt.de/krotik/ecal/util"
  23. )
  24. /*
  25. InbuildFuncMap contains the mapping of inbuild functions.
  26. */
  27. var InbuildFuncMap = map[string]util.ECALFunction{
  28. "range": &rangeFunc{&inbuildBaseFunc{}},
  29. "new": &newFunc{&inbuildBaseFunc{}},
  30. "len": &lenFunc{&inbuildBaseFunc{}},
  31. "del": &delFunc{&inbuildBaseFunc{}},
  32. "add": &addFunc{&inbuildBaseFunc{}},
  33. "concat": &concatFunc{&inbuildBaseFunc{}},
  34. "dumpenv": &dumpenvFunc{&inbuildBaseFunc{}},
  35. "doc": &docFunc{&inbuildBaseFunc{}},
  36. "sleep": &sleepFunc{&inbuildBaseFunc{}},
  37. "raise": &raise{&inbuildBaseFunc{}},
  38. "addEvent": &addevent{&inbuildBaseFunc{}},
  39. "addEventAndWait": &addeventandwait{&addevent{&inbuildBaseFunc{}}},
  40. "setCronTrigger": &setCronTrigger{&inbuildBaseFunc{}},
  41. "setPulseTrigger": &setPulseTrigger{&inbuildBaseFunc{}},
  42. }
  43. /*
  44. inbuildBaseFunc is the base structure for inbuild functions providing some
  45. utility functions.
  46. */
  47. type inbuildBaseFunc struct {
  48. }
  49. /*
  50. AssertNumParam converts a general interface{} parameter into a number.
  51. */
  52. func (ibf *inbuildBaseFunc) AssertNumParam(index int, val interface{}) (float64, error) {
  53. var err error
  54. resNum, ok := val.(float64)
  55. if !ok {
  56. resNum, err = strconv.ParseFloat(fmt.Sprint(val), 64)
  57. if err != nil {
  58. err = fmt.Errorf("Parameter %v should be a number", index)
  59. }
  60. }
  61. return resNum, err
  62. }
  63. /*
  64. AssertMapParam converts a general interface{} parameter into a map.
  65. */
  66. func (ibf *inbuildBaseFunc) AssertMapParam(index int, val interface{}) (map[interface{}]interface{}, error) {
  67. valMap, ok := val.(map[interface{}]interface{})
  68. if ok {
  69. return valMap, nil
  70. }
  71. return nil, fmt.Errorf("Parameter %v should be a map", index)
  72. }
  73. /*
  74. AssertListParam converts a general interface{} parameter into a list.
  75. */
  76. func (ibf *inbuildBaseFunc) AssertListParam(index int, val interface{}) ([]interface{}, error) {
  77. valList, ok := val.([]interface{})
  78. if ok {
  79. return valList, nil
  80. }
  81. return nil, fmt.Errorf("Parameter %v should be a list", index)
  82. }
  83. // Range
  84. // =====
  85. /*
  86. rangeFunc is an interator function which returns a range of numbers.
  87. */
  88. type rangeFunc struct {
  89. *inbuildBaseFunc
  90. }
  91. /*
  92. Run executes this function.
  93. */
  94. func (rf *rangeFunc) Run(instanceID string, vs parser.Scope, is map[string]interface{}, args []interface{}) (interface{}, error) {
  95. var currVal, to float64
  96. var err error
  97. lenargs := len(args)
  98. from := 0.
  99. step := 1.
  100. if lenargs == 0 {
  101. err = fmt.Errorf("Need at least an end range as first parameter")
  102. }
  103. if err == nil {
  104. if stepVal, ok := is[instanceID+"step"]; ok {
  105. step = stepVal.(float64)
  106. from = is[instanceID+"from"].(float64)
  107. to = is[instanceID+"to"].(float64)
  108. currVal = is[instanceID+"currVal"].(float64)
  109. is[instanceID+"currVal"] = currVal + step
  110. // Check for end of iteration
  111. if (from < to && currVal > to) || (from > to && currVal < to) || from == to {
  112. err = util.ErrEndOfIteration
  113. }
  114. } else {
  115. if lenargs == 1 {
  116. to, err = rf.AssertNumParam(1, args[0])
  117. } else {
  118. from, err = rf.AssertNumParam(1, args[0])
  119. if err == nil {
  120. to, err = rf.AssertNumParam(2, args[1])
  121. }
  122. if err == nil && lenargs > 2 {
  123. step, err = rf.AssertNumParam(3, args[2])
  124. }
  125. }
  126. if err == nil {
  127. is[instanceID+"from"] = from
  128. is[instanceID+"to"] = to
  129. is[instanceID+"step"] = step
  130. is[instanceID+"currVal"] = from
  131. currVal = from
  132. }
  133. }
  134. }
  135. if err == nil {
  136. err = util.ErrIsIterator // Identify as iterator
  137. }
  138. return currVal, err
  139. }
  140. /*
  141. DocString returns a descriptive string.
  142. */
  143. func (rf *rangeFunc) DocString() (string, error) {
  144. return "Range function which can be used to iterate over number ranges. Parameters are start, end and step.", nil
  145. }
  146. // New
  147. // ===
  148. /*
  149. newFunc instantiates a new object.
  150. */
  151. type newFunc struct {
  152. *inbuildBaseFunc
  153. }
  154. /*
  155. Run executes this function.
  156. */
  157. func (rf *newFunc) Run(instanceID string, vs parser.Scope, is map[string]interface{}, args []interface{}) (interface{}, error) {
  158. var res interface{}
  159. err := fmt.Errorf("Need a map as first parameter")
  160. if len(args) > 0 {
  161. var argMap map[interface{}]interface{}
  162. if argMap, err = rf.AssertMapParam(1, args[0]); err == nil {
  163. obj := make(map[interface{}]interface{})
  164. res = obj
  165. _, err = rf.addSuperClasses(vs, is, obj, argMap)
  166. if initObj, ok := obj["init"]; ok {
  167. if initFunc, ok := initObj.(*function); ok {
  168. initvs := scope.NewScope(fmt.Sprintf("newfunc: %v", instanceID))
  169. initis := make(map[string]interface{})
  170. _, err = initFunc.Run(instanceID, initvs, initis, args[1:])
  171. }
  172. }
  173. }
  174. }
  175. return res, err
  176. }
  177. /*
  178. addSuperClasses adds super class functions to a given object.
  179. */
  180. func (rf *newFunc) addSuperClasses(vs parser.Scope, is map[string]interface{},
  181. obj map[interface{}]interface{}, template map[interface{}]interface{}) (interface{}, error) {
  182. var err error
  183. var initFunc interface{}
  184. var initSuperList []interface{}
  185. // First loop into the base classes (i.e. top-most classes)
  186. if super, ok := template["super"]; ok {
  187. if superList, ok := super.([]interface{}); ok {
  188. for _, superObj := range superList {
  189. var superInit interface{}
  190. if superTemplate, ok := superObj.(map[interface{}]interface{}); ok {
  191. superInit, err = rf.addSuperClasses(vs, is, obj, superTemplate)
  192. initSuperList = append(initSuperList, superInit) // Build up the list of super functions
  193. }
  194. }
  195. } else {
  196. err = fmt.Errorf("Property _super must be a list of super classes")
  197. }
  198. }
  199. // Copy all properties from template to obj
  200. for k, v := range template {
  201. // Save previous init function
  202. if funcVal, ok := v.(*function); ok {
  203. newFunction := &function{funcVal.name, nil, obj, funcVal.declaration, funcVal.declarationVS}
  204. if k == "init" {
  205. newFunction.super = initSuperList
  206. initFunc = newFunction
  207. }
  208. obj[k] = newFunction
  209. } else {
  210. obj[k] = v
  211. }
  212. }
  213. return initFunc, err
  214. }
  215. /*
  216. DocString returns a descriptive string.
  217. */
  218. func (rf *newFunc) DocString() (string, error) {
  219. return "New creates a new object instance.", nil
  220. }
  221. // Len
  222. // ===
  223. /*
  224. lenFunc returns the size of a list or map.
  225. */
  226. type lenFunc struct {
  227. *inbuildBaseFunc
  228. }
  229. /*
  230. Run executes this function.
  231. */
  232. func (rf *lenFunc) Run(instanceID string, vs parser.Scope, is map[string]interface{}, args []interface{}) (interface{}, error) {
  233. var res float64
  234. err := fmt.Errorf("Need a list or a map as first parameter")
  235. if len(args) > 0 {
  236. argList, ok1 := args[0].([]interface{})
  237. argMap, ok2 := args[0].(map[interface{}]interface{})
  238. if ok1 {
  239. res = float64(len(argList))
  240. err = nil
  241. } else if ok2 {
  242. res = float64(len(argMap))
  243. err = nil
  244. }
  245. }
  246. return res, err
  247. }
  248. /*
  249. DocString returns a descriptive string.
  250. */
  251. func (rf *lenFunc) DocString() (string, error) {
  252. return "Len returns the size of a list or map.", nil
  253. }
  254. // Del
  255. // ===
  256. /*
  257. delFunc removes an element from a list or map.
  258. */
  259. type delFunc struct {
  260. *inbuildBaseFunc
  261. }
  262. /*
  263. Run executes this function.
  264. */
  265. func (rf *delFunc) Run(instanceID string, vs parser.Scope, is map[string]interface{}, args []interface{}) (interface{}, error) {
  266. var res interface{}
  267. err := fmt.Errorf("Need a list or a map as first parameter and an index or key as second parameter")
  268. if len(args) == 2 {
  269. if argList, ok1 := args[0].([]interface{}); ok1 {
  270. var index float64
  271. index, err = rf.AssertNumParam(2, args[1])
  272. if err == nil {
  273. res = append(argList[:int(index)], argList[int(index+1):]...)
  274. }
  275. }
  276. if argMap, ok2 := args[0].(map[interface{}]interface{}); ok2 {
  277. key := fmt.Sprint(args[1])
  278. delete(argMap, key)
  279. res = argMap
  280. err = nil
  281. }
  282. }
  283. return res, err
  284. }
  285. /*
  286. DocString returns a descriptive string.
  287. */
  288. func (rf *delFunc) DocString() (string, error) {
  289. return "Del removes an item from a list or map.", nil
  290. }
  291. // Add
  292. // ===
  293. /*
  294. addFunc adds an element to a list.
  295. */
  296. type addFunc struct {
  297. *inbuildBaseFunc
  298. }
  299. /*
  300. Run executes this function.
  301. */
  302. func (rf *addFunc) Run(instanceID string, vs parser.Scope, is map[string]interface{}, args []interface{}) (interface{}, error) {
  303. var res interface{}
  304. err := fmt.Errorf("Need a list as first parameter and a value as second parameter")
  305. if len(args) > 1 {
  306. var argList []interface{}
  307. if argList, err = rf.AssertListParam(1, args[0]); err == nil {
  308. if len(args) == 3 {
  309. var index float64
  310. if index, err = rf.AssertNumParam(3, args[2]); err == nil {
  311. argList = append(argList, 0)
  312. copy(argList[int(index+1):], argList[int(index):])
  313. argList[int(index)] = args[1]
  314. res = argList
  315. }
  316. } else {
  317. res = append(argList, args[1])
  318. }
  319. }
  320. }
  321. return res, err
  322. }
  323. /*
  324. DocString returns a descriptive string.
  325. */
  326. func (rf *addFunc) DocString() (string, error) {
  327. return "Add adds an item to a list. The item is added at the optionally given index or at the end if no index is specified.", nil
  328. }
  329. // Concat
  330. // ======
  331. /*
  332. concatFunc joins one or more lists together.
  333. */
  334. type concatFunc struct {
  335. *inbuildBaseFunc
  336. }
  337. /*
  338. Run executes this function.
  339. */
  340. func (rf *concatFunc) Run(instanceID string, vs parser.Scope, is map[string]interface{}, args []interface{}) (interface{}, error) {
  341. var res interface{}
  342. err := fmt.Errorf("Need at least two lists as parameters")
  343. if len(args) > 1 {
  344. var argList []interface{}
  345. resList := make([]interface{}, 0)
  346. err = nil
  347. for _, a := range args {
  348. if err == nil {
  349. if argList, err = rf.AssertListParam(1, a); err == nil {
  350. resList = append(resList, argList...)
  351. }
  352. }
  353. }
  354. if err == nil {
  355. res = resList
  356. }
  357. }
  358. return res, err
  359. }
  360. /*
  361. DocString returns a descriptive string.
  362. */
  363. func (rf *concatFunc) DocString() (string, error) {
  364. return "Concat joins one or more lists together. The result is a new list.", nil
  365. }
  366. // dumpenv
  367. // =======
  368. /*
  369. dumpenvFunc returns the current variable environment as a string.
  370. */
  371. type dumpenvFunc struct {
  372. *inbuildBaseFunc
  373. }
  374. /*
  375. Run executes this function.
  376. */
  377. func (rf *dumpenvFunc) Run(instanceID string, vs parser.Scope, is map[string]interface{}, args []interface{}) (interface{}, error) {
  378. return vs.String(), nil
  379. }
  380. /*
  381. DocString returns a descriptive string.
  382. */
  383. func (rf *dumpenvFunc) DocString() (string, error) {
  384. return "Dumpenv returns the current variable environment as a string.", nil
  385. }
  386. // doc
  387. // ===
  388. /*
  389. docFunc returns the docstring of a function.
  390. */
  391. type docFunc struct {
  392. *inbuildBaseFunc
  393. }
  394. /*
  395. Run executes this function.
  396. */
  397. func (rf *docFunc) Run(instanceID string, vs parser.Scope, is map[string]interface{}, args []interface{}) (interface{}, error) {
  398. var res interface{}
  399. err := fmt.Errorf("Need a function as parameter")
  400. if len(args) > 0 {
  401. funcObj, ok := args[0].(util.ECALFunction)
  402. if args[0] == nil {
  403. // Try to lookup by the given identifier
  404. c := is["astnode"].(*parser.ASTNode).Children[0].Children[0]
  405. astring := c.Token.Val
  406. if len(c.Children) > 0 {
  407. astring = fmt.Sprintf("%v.%v", astring, c.Children[0].Token.Val)
  408. }
  409. // Check for stdlib function
  410. if funcObj, ok = stdlib.GetStdlibFunc(astring); !ok {
  411. // Check for inbuild function
  412. funcObj, ok = InbuildFuncMap[astring]
  413. }
  414. }
  415. if ok {
  416. res, err = funcObj.DocString()
  417. }
  418. }
  419. return res, err
  420. }
  421. /*
  422. DocString returns a descriptive string.
  423. */
  424. func (rf *docFunc) DocString() (string, error) {
  425. return "Doc returns the docstring of a function.", nil
  426. }
  427. // sleep
  428. // =====
  429. /*
  430. sleepFunc pauses the current thread for a number of micro seconds.
  431. */
  432. type sleepFunc struct {
  433. *inbuildBaseFunc
  434. }
  435. /*
  436. Run executes this function.
  437. */
  438. func (rf *sleepFunc) Run(instanceID string, vs parser.Scope, is map[string]interface{}, args []interface{}) (interface{}, error) {
  439. var res interface{}
  440. err := fmt.Errorf("Need number of micro seconds as parameter")
  441. if len(args) > 0 {
  442. var micros float64
  443. micros, err = rf.AssertNumParam(1, args[0])
  444. if err == nil {
  445. time.Sleep(time.Duration(micros) * time.Microsecond)
  446. }
  447. }
  448. return res, err
  449. }
  450. /*
  451. DocString returns a descriptive string.
  452. */
  453. func (rf *sleepFunc) DocString() (string, error) {
  454. return "Sleep pauses the current thread for a number of micro seconds.", nil
  455. }
  456. // raise
  457. // =====
  458. /*
  459. raise returns an error. Outside of sinks this will stop the code execution
  460. if the error is not handled by try / except. Inside a sink only the specific sink
  461. will fail. This error can be used to break trigger sequences of sinks if
  462. FailOnFirstErrorInTriggerSequence is set.
  463. */
  464. type raise struct {
  465. *inbuildBaseFunc
  466. }
  467. /*
  468. Run executes this function.
  469. */
  470. func (rf *raise) Run(instanceID string, vs parser.Scope, is map[string]interface{}, args []interface{}) (interface{}, error) {
  471. var err error
  472. var detailMsg string
  473. var detail interface{}
  474. if len(args) > 0 {
  475. err = fmt.Errorf("%v", args[0])
  476. if len(args) > 1 {
  477. if args[1] != nil {
  478. detailMsg = fmt.Sprint(args[1])
  479. }
  480. if len(args) > 2 {
  481. detail = args[2]
  482. }
  483. }
  484. }
  485. erp := is["erp"].(*ECALRuntimeProvider)
  486. node := is["astnode"].(*parser.ASTNode)
  487. return nil, &util.RuntimeErrorWithDetail{
  488. RuntimeError: erp.NewRuntimeError(err, detailMsg, node).(*util.RuntimeError),
  489. Environment: vs,
  490. Data: detail,
  491. }
  492. }
  493. /*
  494. DocString returns a descriptive string.
  495. */
  496. func (rf *raise) DocString() (string, error) {
  497. return "Raise returns an error object.", nil
  498. }
  499. // addEvent
  500. // ========
  501. /*
  502. addevent adds an event to trigger sinks. This function will return immediately
  503. and not wait for the event cascade to finish. Use this function for event cascades.
  504. */
  505. type addevent struct {
  506. *inbuildBaseFunc
  507. }
  508. /*
  509. Run executes this function.
  510. */
  511. func (rf *addevent) Run(instanceID string, vs parser.Scope, is map[string]interface{}, args []interface{}) (interface{}, error) {
  512. return rf.addEvent(func(proc engine.Processor, event *engine.Event, scope *engine.RuleScope) (interface{}, error) {
  513. var monitor engine.Monitor
  514. parentMonitor, ok := is["monitor"]
  515. if scope != nil || !ok {
  516. monitor = proc.NewRootMonitor(nil, scope)
  517. } else {
  518. monitor = parentMonitor.(engine.Monitor).NewChildMonitor(0)
  519. }
  520. _, err := proc.AddEvent(event, monitor)
  521. return nil, err
  522. }, is, args)
  523. }
  524. func (rf *addevent) addEvent(addFunc func(engine.Processor, *engine.Event, *engine.RuleScope) (interface{}, error),
  525. is map[string]interface{}, args []interface{}) (interface{}, error) {
  526. var res interface{}
  527. var stateMap map[interface{}]interface{}
  528. erp := is["erp"].(*ECALRuntimeProvider)
  529. proc := erp.Processor
  530. if proc.Stopped() {
  531. proc.Start()
  532. }
  533. err := fmt.Errorf("Need at least three parameters: name, kind and state")
  534. if len(args) > 2 {
  535. if stateMap, err = rf.AssertMapParam(3, args[2]); err == nil {
  536. var scope *engine.RuleScope
  537. event := engine.NewEvent(
  538. fmt.Sprint(args[0]),
  539. strings.Split(fmt.Sprint(args[1]), "."),
  540. stateMap,
  541. )
  542. if len(args) > 3 {
  543. var scopeMap map[interface{}]interface{}
  544. // Add optional scope - if not specified it is { "": true }
  545. if scopeMap, err = rf.AssertMapParam(4, args[3]); err == nil {
  546. var scopeData = map[string]bool{}
  547. for k, v := range scopeMap {
  548. b, _ := strconv.ParseBool(fmt.Sprint(v))
  549. scopeData[fmt.Sprint(k)] = b
  550. }
  551. scope = engine.NewRuleScope(scopeData)
  552. }
  553. }
  554. if err == nil {
  555. res, err = addFunc(proc, event, scope)
  556. }
  557. }
  558. }
  559. return res, err
  560. }
  561. /*
  562. DocString returns a descriptive string.
  563. */
  564. func (rf *addevent) DocString() (string, error) {
  565. return "AddEvent adds an event to trigger sinks. This function will return " +
  566. "immediately and not wait for the event cascade to finish.", nil
  567. }
  568. // addEventAndWait
  569. // ===============
  570. /*
  571. addeventandwait adds an event to trigger sinks. This function will return once
  572. the event cascade has finished and return all errors.
  573. */
  574. type addeventandwait struct {
  575. *addevent
  576. }
  577. /*
  578. Run executes this function.
  579. */
  580. func (rf *addeventandwait) Run(instanceID string, vs parser.Scope, is map[string]interface{}, args []interface{}) (interface{}, error) {
  581. return rf.addEvent(func(proc engine.Processor, event *engine.Event, scope *engine.RuleScope) (interface{}, error) {
  582. var res []interface{}
  583. rm := proc.NewRootMonitor(nil, scope)
  584. m, err := proc.AddEventAndWait(event, rm)
  585. if m != nil {
  586. allErrors := m.(*engine.RootMonitor).AllErrors()
  587. for _, e := range allErrors {
  588. errors := map[interface{}]interface{}{}
  589. for k, v := range e.ErrorMap {
  590. se := v.(*util.RuntimeErrorWithDetail)
  591. // Note: The variable scope of the sink (se.environment)
  592. // was also captured - for now it is not exposed to the
  593. // language environment
  594. errors[k] = map[interface{}]interface{}{
  595. "error": se.Error(),
  596. "type": se.Type.Error(),
  597. "detail": se.Detail,
  598. "data": se.Data,
  599. }
  600. }
  601. item := map[interface{}]interface{}{
  602. "event": map[interface{}]interface{}{
  603. "name": e.Event.Name(),
  604. "kind": strings.Join(e.Event.Kind(), "."),
  605. "state": e.Event.State(),
  606. },
  607. "errors": errors,
  608. }
  609. res = append(res, item)
  610. }
  611. }
  612. return res, err
  613. }, is, args)
  614. }
  615. /*
  616. DocString returns a descriptive string.
  617. */
  618. func (rf *addeventandwait) DocString() (string, error) {
  619. return "AddEventAndWait adds an event to trigger sinks. This function will " +
  620. "return once the event cascade has finished.", nil
  621. }
  622. // setCronTrigger
  623. // ==============
  624. /*
  625. setCronTrigger adds a periodic cron job which fires events.
  626. */
  627. type setCronTrigger struct {
  628. *inbuildBaseFunc
  629. }
  630. /*
  631. Run executes this function.
  632. */
  633. func (ct *setCronTrigger) Run(instanceID string, vs parser.Scope, is map[string]interface{}, args []interface{}) (interface{}, error) {
  634. var res interface{}
  635. err := fmt.Errorf("Need a cronspec, an event name and an event scope as parameters")
  636. if len(args) > 2 {
  637. var cs *timeutil.CronSpec
  638. cronspec := fmt.Sprint(args[0])
  639. eventname := fmt.Sprint(args[1])
  640. eventkind := strings.Split(fmt.Sprint(args[2]), ".")
  641. erp := is["erp"].(*ECALRuntimeProvider)
  642. proc := erp.Processor
  643. if proc.Stopped() {
  644. proc.Start()
  645. }
  646. if cs, err = timeutil.NewCronSpec(cronspec); err == nil {
  647. res = cs.String()
  648. tick := 0
  649. erp.Cron.RegisterSpec(cs, func() {
  650. tick += 1
  651. now := erp.Cron.NowFunc()
  652. event := engine.NewEvent(eventname, eventkind, map[interface{}]interface{}{
  653. "time": now,
  654. "timestamp": fmt.Sprintf("%d", now.UnixNano()/int64(time.Millisecond)),
  655. "tick": float64(tick),
  656. })
  657. monitor := proc.NewRootMonitor(nil, nil)
  658. _, err := proc.AddEvent(event, monitor)
  659. errorutil.AssertTrue(err == nil,
  660. fmt.Sprintf("Could not add cron event for trigger %v %v %v: %v",
  661. cronspec, eventname, eventkind, err))
  662. })
  663. }
  664. }
  665. return res, err
  666. }
  667. /*
  668. DocString returns a descriptive string.
  669. */
  670. func (ct *setCronTrigger) DocString() (string, error) {
  671. return "setCronTrigger adds a periodic cron job which fires events.", nil
  672. }
  673. // setPulseTrigger
  674. // ==============
  675. /*
  676. setPulseTrigger adds recurring events in very short intervals.
  677. */
  678. type setPulseTrigger struct {
  679. *inbuildBaseFunc
  680. }
  681. /*
  682. Run executes this function.
  683. */
  684. func (pt *setPulseTrigger) Run(instanceID string, vs parser.Scope, is map[string]interface{}, args []interface{}) (interface{}, error) {
  685. err := fmt.Errorf("Need micro second interval, an event name and an event scope as parameters")
  686. if len(args) > 2 {
  687. var micros float64
  688. micros, err = pt.AssertNumParam(1, args[0])
  689. if err == nil {
  690. eventname := fmt.Sprint(args[1])
  691. eventkind := strings.Split(fmt.Sprint(args[2]), ".")
  692. erp := is["erp"].(*ECALRuntimeProvider)
  693. proc := erp.Processor
  694. if proc.Stopped() {
  695. proc.Start()
  696. }
  697. tick := 0
  698. go func() {
  699. var lastmicros int64
  700. for {
  701. time.Sleep(time.Duration(micros) * time.Microsecond)
  702. tick += 1
  703. now := time.Now()
  704. micros := now.UnixNano() / int64(time.Microsecond)
  705. event := engine.NewEvent(eventname, eventkind, map[interface{}]interface{}{
  706. "currentMicros": float64(micros),
  707. "lastMicros": float64(lastmicros),
  708. "timestamp": fmt.Sprintf("%d", now.UnixNano()/int64(time.Microsecond)),
  709. "tick": float64(tick),
  710. })
  711. lastmicros = micros
  712. monitor := proc.NewRootMonitor(nil, nil)
  713. _, err := proc.AddEvent(event, monitor)
  714. if proc.Stopped() {
  715. break
  716. }
  717. errorutil.AssertTrue(err == nil,
  718. fmt.Sprintf("Could not add pulse event for trigger %v %v %v: %v",
  719. micros, eventname, eventkind, err))
  720. }
  721. }()
  722. }
  723. }
  724. return nil, err
  725. }
  726. /*
  727. DocString returns a descriptive string.
  728. */
  729. func (pt *setPulseTrigger) DocString() (string, error) {
  730. return "setPulseTrigger adds recurring events in microsecond intervals.", nil
  731. }