func_provider.go 23 KB

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