func_provider.go 23 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040
  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. se := v.(*util.RuntimeErrorWithDetail)
  678. // Note: The variable scope of the sink (se.environment)
  679. // was also captured - for now it is not exposed to the
  680. // language environment
  681. errors[k] = map[interface{}]interface{}{
  682. "error": se.Error(),
  683. "type": se.Type.Error(),
  684. "detail": se.Detail,
  685. "data": se.Data,
  686. }
  687. }
  688. item := map[interface{}]interface{}{
  689. "event": map[interface{}]interface{}{
  690. "name": e.Event.Name(),
  691. "kind": strings.Join(e.Event.Kind(), "."),
  692. "state": e.Event.State(),
  693. },
  694. "errors": errors,
  695. }
  696. res = append(res, item)
  697. }
  698. }
  699. return res, err
  700. }, is, args)
  701. }
  702. /*
  703. DocString returns a descriptive string.
  704. */
  705. func (rf *addeventandwait) DocString() (string, error) {
  706. return "AddEventAndWait adds an event to trigger sinks. This function will " +
  707. "return once the event cascade has finished.", nil
  708. }
  709. // setCronTrigger
  710. // ==============
  711. /*
  712. setCronTrigger adds a periodic cron job which fires events.
  713. */
  714. type setCronTrigger struct {
  715. *inbuildBaseFunc
  716. }
  717. /*
  718. Run executes this function.
  719. */
  720. func (ct *setCronTrigger) Run(instanceID string, vs parser.Scope, is map[string]interface{}, tid uint64, args []interface{}) (interface{}, error) {
  721. var res interface{}
  722. err := fmt.Errorf("Need a cronspec, an event name and an event scope as parameters")
  723. if len(args) > 2 {
  724. var cs *timeutil.CronSpec
  725. cronspec := fmt.Sprint(args[0])
  726. eventname := fmt.Sprint(args[1])
  727. eventkind := strings.Split(fmt.Sprint(args[2]), ".")
  728. erp := is["erp"].(*ECALRuntimeProvider)
  729. proc := erp.Processor
  730. if proc.Stopped() {
  731. proc.Start()
  732. }
  733. if cs, err = timeutil.NewCronSpec(cronspec); err == nil {
  734. res = cs.String()
  735. tick := 0
  736. erp.Cron.RegisterSpec(cs, func() {
  737. tick++
  738. now := erp.Cron.NowFunc()
  739. event := engine.NewEvent(eventname, eventkind, map[interface{}]interface{}{
  740. "time": now,
  741. "timestamp": fmt.Sprintf("%d", now.UnixNano()/int64(time.Millisecond)),
  742. "tick": float64(tick),
  743. })
  744. monitor := proc.NewRootMonitor(nil, nil)
  745. _, err := proc.AddEvent(event, monitor)
  746. if status := proc.Status(); status != "Stopped" && status != "Stopping" {
  747. errorutil.AssertTrue(err == nil,
  748. fmt.Sprintf("Could not add cron event for trigger %v %v %v: %v",
  749. cronspec, eventname, eventkind, err))
  750. }
  751. })
  752. }
  753. }
  754. return res, err
  755. }
  756. /*
  757. DocString returns a descriptive string.
  758. */
  759. func (ct *setCronTrigger) DocString() (string, error) {
  760. return "setCronTrigger adds a periodic cron job which fires events.", nil
  761. }
  762. // setPulseTrigger
  763. // ==============
  764. /*
  765. setPulseTrigger adds recurring events in very short intervals.
  766. */
  767. type setPulseTrigger struct {
  768. *inbuildBaseFunc
  769. }
  770. /*
  771. Run executes this function.
  772. */
  773. func (pt *setPulseTrigger) Run(instanceID string, vs parser.Scope, is map[string]interface{}, tid uint64, args []interface{}) (interface{}, error) {
  774. err := fmt.Errorf("Need micro second interval, an event name and an event scope as parameters")
  775. if len(args) > 2 {
  776. var micros float64
  777. micros, err = pt.AssertNumParam(1, args[0])
  778. if err == nil {
  779. eventname := fmt.Sprint(args[1])
  780. eventkind := strings.Split(fmt.Sprint(args[2]), ".")
  781. erp := is["erp"].(*ECALRuntimeProvider)
  782. proc := erp.Processor
  783. if proc.Stopped() {
  784. proc.Start()
  785. }
  786. tick := 0
  787. go func() {
  788. var lastmicros int64
  789. for {
  790. time.Sleep(time.Duration(micros) * time.Microsecond)
  791. tick++
  792. now := time.Now()
  793. micros := now.UnixNano() / int64(time.Microsecond)
  794. event := engine.NewEvent(eventname, eventkind, map[interface{}]interface{}{
  795. "currentMicros": float64(micros),
  796. "lastMicros": float64(lastmicros),
  797. "timestamp": fmt.Sprintf("%d", now.UnixNano()/int64(time.Microsecond)),
  798. "tick": float64(tick),
  799. })
  800. lastmicros = micros
  801. monitor := proc.NewRootMonitor(nil, nil)
  802. _, err := proc.AddEvent(event, monitor)
  803. if status := proc.Status(); status == "Stopped" || status == "Stopping" {
  804. break
  805. }
  806. errorutil.AssertTrue(err == nil,
  807. fmt.Sprintf("Could not add pulse event for trigger %v %v %v: %v",
  808. micros, eventname, eventkind, err))
  809. }
  810. }()
  811. }
  812. }
  813. return nil, err
  814. }
  815. /*
  816. DocString returns a descriptive string.
  817. */
  818. func (pt *setPulseTrigger) DocString() (string, error) {
  819. return "setPulseTrigger adds recurring events in microsecond intervals.", nil
  820. }