func_provider.go 23 KB

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