func (self *json_outstream) Initialize() kernel.Error {
	self.MsgDebug("== initialize ==\n")
	if !self.Algorithm.Initialize().IsSuccess() {
		self.MsgError("could not initialize base-class\n")
		return kernel.StatusCode(1)
	}

	self.item_names = self.GetProperty("Items").([]string)

	fname := self.GetProperty("Output").(string)
	self.MsgInfo("output file: [%v]\n", fname)
	self.MsgInfo("items: %v\n", self.item_names)

	svcloc := kernel.GetSvcLocator()
	if svcloc == nil {
		self.MsgError("could not retrieve service locator !\n")
		return kernel.StatusCode(1)
	}
	svc := svcloc.GetService("outstreamsvc").(kernel.IOutputStreamSvc)
	if svc == nil {
		self.MsgError("could not retrieve [outstreamsvc] !\n")
		return kernel.StatusCode(1)
	}

	self.handle = svc.NewOutputStream(fname)
	if self.handle == nil {
		self.MsgError("could not retrieve json output stream [%s] !\n", fname)
		return kernel.StatusCode(1)
	}

	return kernel.StatusCode(0)
}

func (self *alg1) Initialize() kernel.Error {
	self.MsgInfo("== initialize ==\n")
	if !self.Algorithm.Initialize().IsSuccess() {
		self.MsgError("could not initialize base-class\n")
		return kernel.StatusCode(1)
	}
	return kernel.StatusCode(0)
}

func (self *svc2) InitializeSvc() kernel.Error {
	self.MsgInfo("~~ initialize ~~\n")
	if !self.Service.InitializeSvc().IsSuccess() {
		self.MsgError("could not initialize base-class\n")
		return kernel.StatusCode(1)
	}
	return kernel.StatusCode(0)
}

func (self *tool1) InitializeTool() kernel.Error {
	self.MsgInfo("~~ initialize ~~\n")
	if !self.AlgTool.InitializeTool().IsSuccess() {
		self.MsgError("could not initialize base-class\n")
		return kernel.StatusCode(1)
	}
	return kernel.StatusCode(0)
}

func (self *evtproc) mp_NextEvent(evtmax int) kernel.Error {

	handle := func(evt *evtstate, out_queue chan<- *evtstate) {
		self.MsgInfo("nextEvent[%v]...\n", evt.idx)
		evt.sc = self.ExecuteEvent(evt)
		out_queue <- evt
	}

	serve_evts := func(in_evt_queue <-chan *evtstate, out_evt_queue chan<- *evtstate, quit <-chan bool) {
		for {
			select {
			case ievt := <-in_evt_queue:
				go handle(ievt, out_evt_queue)
			case <-quit:
				//println("quit requested !")
				return
			}
		}
	}

	start_evt_server := func(nworkers int) (in_evt_queue,
		out_evt_queue chan *evtstate,
		quit chan bool) {
		in_evt_queue = make(chan *evtstate, nworkers)
		out_evt_queue = make(chan *evtstate)
		quit = make(chan bool)
		go serve_evts(in_evt_queue, out_evt_queue, quit)
		return in_evt_queue, out_evt_queue, quit
	}

	in_evt_queue, out_evt_queue, quit := start_evt_server(self.nworkers)
	//self.MsgInfo("sending requests...\n")
	for i := 0; i < evtmax; i++ {
		in_evt_queue <- new_evtstate(i)
	}
	//self.MsgInfo("sending requests... [done]\n")
	n_fails := 0
	n_processed := 0
	for evt := range out_evt_queue {
		//self.MsgDebug("out-evt-queue: %v %v\n",evt.idx, evt.sc)
		if !evt.sc.IsSuccess() {
			n_fails++
		}
		n_processed++
		if n_processed == evtmax {
			quit <- true
			close(out_evt_queue)
			//self.MsgDebug("closing evt server...\n")
			break
		}
	}
	if n_fails != 0 {
		return kernel.StatusCode(1)
	}
	return kernel.StatusCode(0)
}

func (self *json_outstream_svc) InitializeSvc() kernel.Error {

	self.MsgDebug("== initialize ==\n")
	if !self.Service.InitializeSvc().IsSuccess() {
		self.MsgError("could not initialize base-class\n")
		return kernel.StatusCode(1)
	}

	return kernel.StatusCode(0)
}

func (self *evtproc) seq_NextEvent(evtmax int) kernel.Error {

	self.MsgInfo("nextEvent[%v]...\n", evtmax)
	for i := 0; i < evtmax; i++ {
		ctx := new_evtstate(i)
		if !self.ExecuteEvent(ctx).IsSuccess() {
			self.MsgError("failed to execute evt idx %03v\n", i)
			return kernel.StatusCode(1)
		}
	}
	return kernel.StatusCode(0)
}

func (self *evtproc) ExecuteEvent(ictx kernel.IEvtCtx) kernel.Error {
	ctx := ictx.Idx()
	self.MsgDebug("executing event [%v]... (#algs: %v)\n", ctx, len(self.algs))
	for i, alg := range self.algs {
		self.MsgDebug("-- ctx:%03v --> [%s]...\n", ctx, alg.CompName())
		if !alg.Execute(ictx).IsSuccess() {
			self.MsgError("pb executing alg #%v (%s) for ctx:%v\n",
				i, alg.CompName(), ictx.Idx())
			return kernel.StatusCode(1)
		}
	}
	self.MsgDebug("data: %v\n", ictx.Store())
	return kernel.StatusCode(0)
}

func new_evtstate(idx int) *evtstate {
	self := &evtstate{}
	self.idx = idx
	self.sc = kernel.StatusCode(0)
	self.data = make(kernel.DataStore)
	return self
}

func (self *json_outstream_handle) Close() kernel.Error {

	self.quit <- true

	msg := self.svc.(kernel.IMessager)
	msg.MsgDebug("--> closing json-handle [%v]\n", self.w.Name())

	// _,ok = <-self.quit
	// if ok {
	// 	self.quit <- true
	// }
	close(self.quit)
	close(self.errs)
	close(self.data)

	fd := self.w.Fd()
	if fd >= 0 {
		fname := self.w.Name()
		err := self.w.Close()
		if err != nil {
			msg.MsgError("closing fd: [%v] name [%v]. err: %v\n",
				fd, fname, err)
			return kernel.StatusCodeWithErr(1, err)
		}
	}
	return kernel.StatusCode(0)
}

func (self *alg_adder) Execute(ctx kernel.IEvtCtx) kernel.Error {
	self.MsgDebug("== execute == [%v]\n", ctx.Idx())

	njets := 1 + ctx.Idx()
	val := self.val + 1
	store := self.EvtStore(ctx)

	if store.Has("njets") {
		njets += store.Get("njets").(int)
	}
	store.Put("njets", njets)

	if store.Has("ptjets") {
		val += store.Get("ptjets").(float64)
	}
	store.Put("ptjets", val)

	cnt := &simple_counter{0}
	if store.Has(self.cnt_key) {
		cnt = store.Get(self.cnt_key).(*simple_counter)
	}
	cnt.Cnt += 1
	store.Put(self.cnt_key, cnt)
	return kernel.StatusCode(0)
}

func (self *json_outstream_handle) Write(data interface{}) kernel.Error {
	/* // FIXME: how to get the error back ??
	err := self.enc.Encode(data)
	if err != nil {
		return kernel.StatusCodeWithErr(1, err)
	}
	*/
	self.data <- data
	select {
	case err := <-self.errs:
		if err != nil {
			msg := self.svc.(kernel.IMessager)
			msg.MsgError("--> write got: %v\n", err)
			return kernel.StatusCodeWithErr(1, err)
		}
	default:
		return kernel.StatusCode(0)
	}
	return kernel.StatusCode(0)
}

func (self *gob_outstream) Execute(ctx kernel.IEvtCtx) kernel.Error {
	self.MsgDebug("== execute ==\n")
	store := self.EvtStore(ctx)
	if store == nil {
		self.MsgError("could not retrieve evt-store\n")
	}
	var err error
	allgood := true
	for _, k := range self.item_names {
		err = self.enc.Encode(store.Get(k))
		if err != nil {
			self.MsgError("error while writing store content at [%v]: %v\n",
				k, err)
			allgood = false
		}
	}

	/*
		hdr_offset := 1
		val := make([]interface{}, len(self.item_names)+hdr_offset)
		val[0] = ctx.Idx()

		for i,k := range self.item_names {
			val[i+hdr_offset] = store.Get(k)
		}
		for idx,v := range val {
			err = self.enc.Encode(val)
			if err != nil {
				self.MsgError("error while encoding data [%v|%v]: %v\n",
					idx, v, err)
				allgood = false
			}
		}
	*/

	if !allgood {
		return kernel.StatusCode(1)
	}
	return kernel.StatusCode(0)
}

func (self *evtproc) InitializeSvc() kernel.Error {

	sc := self.Service.InitializeSvc()
	if !sc.IsSuccess() {
		return sc
	}
	self.nworkers = self.GetProperty("NbrWorkers").(int)
	//self.nworkers = 2
	self.MsgInfo("n-workers: %v\n", self.nworkers)
	svcloc := kernel.GetSvcLocator()
	if svcloc == nil {
		self.MsgError("could not retrieve ISvcLocator !\n")
		return kernel.StatusCode(1)
	}
	appmgr := svcloc.(kernel.IComponentMgr).GetComp("app-mgr")
	if appmgr == nil {
		self.MsgError("could not retrieve 'app-mgr'\n")
	}
	propmgr := appmgr.(kernel.IProperty)
	alg_names := propmgr.GetProperty("Algs").([]string)
	self.MsgInfo("got alg-names: %v\n", len(alg_names))

	if len(alg_names) > 0 {
		comp_mgr := appmgr.(kernel.IComponentMgr)
		self.algs = make([]kernel.IAlgorithm, len(alg_names))
		for i, alg_name := range alg_names {
			ialg, isalg := comp_mgr.GetComp(alg_name).(kernel.IAlgorithm)
			if isalg {
				self.algs[i] = ialg
			}
		}
	}
	self.MsgInfo("got alg-list: %v\n", len(self.algs))

	return kernel.StatusCode(0)
}

func (self *evtproc) test_0() {
	handle := func(queue <-chan int) kernel.Error {
		sc := kernel.StatusCode(0)
		for i := range queue {
			ctx := new_evtstate(i)
			self.MsgInfo("   --> handling [%i]...\n", i)
			sc = self.ExecuteEvent(ctx)
		}
		return sc
	}

	max_in_flight := 4
	serve := func(queue <-chan int, quit <-chan bool) kernel.Error {
		for i := 0; i < max_in_flight; i++ {
			go handle(queue)
		}
		<-quit // wait to be told to exit
		return kernel.StatusCode(0)
	}

	quit := make(chan bool)

	self.MsgInfo("-- filling the event queue...\n")
	queue := make(chan int)
	go func() {
		for i := 0; i < 20; i++ {
			queue <- i
		}
	}()
	self.MsgInfo("-- starting to serve 20 events...\n")
	go serve(queue, quit)
	self.MsgInfo("-- requests sent...\n")
	time.Sleep(2000000000)
	quit <- true
	self.MsgInfo("-- done.\n")
}

func (self *json_outstream_svc) FinalizeSvc() kernel.Error {

	self.MsgDebug("== finalize ==\n")
	if !self.Service.FinalizeSvc().IsSuccess() {
		self.MsgError("could not finalize base-class\n")
		return kernel.StatusCode(1)
	}

	self.MsgDebug("== closing output streams...\n")
	allgood := true
	for n, stream := range self.streams {
		self.MsgDebug("-- closing [%s]...\n", n)
		sc := stream.Close()
		if !sc.IsSuccess() {
			self.MsgError("problem closing stream [%s]: %v\n", n, sc)
			allgood = false
		}
	}
	if !allgood {
		return kernel.StatusCode(1)
	}

	return kernel.StatusCode(0)
}

func (self *evtproc) StopRun() kernel.Error {
	self.MsgInfo("stopping run...\n")
	return kernel.StatusCode(0)
}

func (self *gob_outstream) Finalize() kernel.Error {
	self.MsgDebug("== finalize ==\n")
	//self.w.Close()

	return kernel.StatusCode(0)
}

func (self *svc2) FinalizeSvc() kernel.Error {
	self.MsgInfo("~~ finalize ~~\n")
	return kernel.StatusCode(0)
}

func (self *datastore) ClearStore() kernel.Error {
	for k, _ := range self.store {
		delete(self.store, k)
	}
	return kernel.StatusCode(0)
}