Einführung

Hallo. Yorozu Berater Sugimon: lecker :. Dieses Mal werde ich versuchen, eine Komponente (Adapter) mit dem SDK von HULFT IoT Edge Streaming zu entwickeln. Es wird so beschrieben, dass es in allen drei Teilen abgeschlossen wird.

Was ist diesmal zu tun?

Dieses Mal werde ich als zweiten Schritt einen Plug mit dem HULFT IoT EdgeStreaming Plugin SDK entwickeln. Es gibt weitere Artikel zum Einrichten und Ausführen. Lesen Sie daher diese.

: arrow_forward: [Ich habe versucht, ein Plug-In mit HULFT IoT Edge Streaming zu erstellen Setup] (https://qiita.com/sugimon/items/cf7503479c6e4c46c3b3) : arrow_forward: Ich habe versucht, ein Plug-In mit HULFT IoT Edge Streaming Entwicklung zu erstellen. : arrow_forward: [Ich habe versucht, ein Plug-In mit HULFT IoT Edge Streaming Ausführung zu erstellen](https://qiita.com/sugimon/items/93f47d7bd472a8b18e54)

Übersicht über das Erstellen eines Plug-Ins

Die Konfiguration zum Erstellen eines Plug-Ins für EdgeStreaming lautet wie folgt.

Konfiguration der Ausführungsumgebung

Dies ist der Runtime-Teil, der den von Studio angegebenen Streaming-Prozess ausführt.

SourceOperation `Dies ist eine Operation zum Generieren von Streaming-Daten (Tupel). Es wird "Eingabeverarbeitung" sein.

JSON-Format `

SinkOperation `Dies ist eine Operation zum Ausgeben von Streaming-Daten (Tupel) nach außen. Es wird "Ausgabeverarbeitung". ``

UDSFOperation Dies ist eine Operation zum Konvertieren von Streaming-Daten (Tupel). Es gibt auch das konvertierte Streming (Tupel) aus. "Eingabe / Ausgabe (Konvertierung) Verarbeitung"

Konfiguration der Entwicklungsumgebung

Generiert eine BQL (SQL-ähnliche Syntax), die RunTime mitteilt, wie die Streaming-Verarbeitung durchgeführt werden soll.

AdapterModuleComponent `Eine Klasse, die eine Komponente (Adapter) darstellt. Es ist viele zu eins mit der Operation. ``

BQLPluginSourceOperationFactory `Definiert die Eigenschaften von SourceOperartion. "Eingabeverarbeitung" Gibt die BQL (Create Source-Anweisung) aus, die die Source-Operation erstellt. ``

BQLPluginOutputOperationFactory `Definiert die Eigenschaften von SinkOperation. "Ausgabeverarbeitung" Gibt eine BQL (Create Sink-Anweisung) aus, die eine Sink-Operation erstellt. ``

BQLPluginUDSFOperationFactory `Definiert UDSF-Eigenschaften. "Eingabe / Ausgabe (Konvertierung) Verarbeitung" Gibt eine BQL (Select-Anweisung) aus, die eine UDSF-Operation generiert. ``

Plug-In erstellen

Dieses Mal habe ich versucht, ein Plug-In in der folgenden Form zu erstellen.

·Grundinformation

** ・ Betriebsinformationen **

Implementierung auf der Seite der Ausführungsumgebung (Golang)

Implementiert den von Studio angegebenen Streaming-Prozess.

** Umweltvorbereitung **

Erstellen eines Modulverzeichnisses $ $ SDK_HOME / dev / go / src / github.sis.saison.co.jp / sherpa / es-agent / sample

Die Quelldatei wird hier gespeichert. (source.go, sink.go, udsf.go)

Erstellen Sie ein Plugin-Verzeichnis ・ $ SDK_HOME / dev / go / src / github.sis.saison.co.jp / sherpa / es-agent / sample / plugin
Erstellen Sie ein externes Verzeichnis $ $ SDK_HOME / dev / go / src / github.sis.saison.co.jp / sherpa / es-agent / sample / external

** Erstellen einer Quelldatei ** Erstellen wir eine Quelldatei mit dem folgenden Dateinamen im Modulverzeichnis (Beispiel in diesem Fall).

-Die Dateistruktur ist wie folgt.

├─sample
│   ├─source.go (Eingabeverarbeitung)
│   ├─sink.go (Ausgabeverarbeitung)
│   ├─udsf.go (Eingabe / Ausgabe-Konvertierungsverarbeitung)
│   │ 
│   ├─external
│   │   ├─plugin_main.go (Hauptfunktion)
│   │   │ 
│   └─plugin
│   │   ├─plugin.go (registriere die Quelldateien unter jedem Beispiel)
│   │   │

Nun erstellen wir die Quelldatei.

** ・ source.go (Eingabeverarbeitung) ** Erstellen Sie einen Prozess, der in regelmäßigen Zeitintervallen Pseudozufallszahlen generiert. Der Hauptfluss ist wie folgt.

Empfangen Sie den Integer-Typ-Parameter "Intervall".
Generieren Sie Pseudozufallszahlen in Intervallen
Generieren Sie ein Tupel mit JSON-Schemadaten

`Tuple`


{
  "type": "object",
  "required": ["payload"],
  "properties": {
    "payload": {
      "type": "object",
      "required": ["value"],
      "properties": {
        "value": {
          "type": "number"
        }
      }
    }
  }
}

Die ausgegebenen Tupel-JSON-Daten haben das folgende Format.

{"payload": {"value": 3.5423242}}

Ich habe source.go wie folgt erstellt.

`source.go`


package sample

import (
	"math/rand"
	"time"

	"gopkg.in/sensorbee/sensorbee.v0/bql"
	"gopkg.in/sensorbee/sensorbee.v0/core"
	"gopkg.in/sensorbee/sensorbee.v0/data"
)

type source struct {
	interval time.Duration
	term     chan struct{}
}

func (s *source) GenerateStream(ctx *core.Context, w core.Writer) error {
	rand.Seed(time.Now().UnixNano())

	next := time.Now()
	for {
		val := rand.Float64()
		m := data.Map{"value": data.Float(val)}
		t := core.NewTuple(data.Map{"payload": m})
		if s.interval > 0 {
			t.Timestamp = next
		}
		ctx.Log().Debug("generation: ", val)
		if err := w.Write(ctx, t); err != nil {
			return err
		}

		if s.interval > 0 {
			now := time.Now()
			next = next.Add(s.interval)
			if next.Before(now) {
				next = now.Add(s.interval)
			}

			select {
			case <-s.term:
				return core.ErrSourceStopped
			case <-time.After(next.Sub(now)):
			}
		}
	}
	return nil
}

func (s *source) Stop(ctx *core.Context) error {
	s.term <- struct{}{}
	return nil
}

func CreateSource(ctx *core.Context, ioParams *bql.IOParams, params data.Map) (core.Source, error) {
	interval, err := getInterval(params)
	if err != nil {
		return nil, err
	}

	return &source{
		interval: interval,
		term:     make(chan struct{}),
	}, nil
}

func getInterval(params data.Map) (time.Duration, error) {
	interval := 1 * time.Second
	if v, ok := params["interval"]; ok {
		i, err := data.ToDuration(v)
		if err != nil {
			return interval, err
		}
		interval = i
	}
	return interval, nil
}

** ・ sink.go (Ausgabeverarbeitung) ** Erstellen Sie einen Prozess, der nach der Anzahl der effektiven Dezimalstellen abgeschnitten und in das Protokoll ausgegeben wird.

Der Hauptfluss ist wie folgt.

Empfangen Sie den Integer-Typ-Parameter "decimal".
Geben Sie den empfangenen Wert mit der Anzahl der gültigen Dezimalstellen "Dezimal" an die Standardausgabe aus.
Tupel mit Schemadaten von JSON-Daten empfangen`

`Tuple`


{
  "type": "object",
  "required": ["payload"],
  "properties": {
    "payload": {
      "type": "object",
      "required": ["value", "formula"],
      "properties": {
        "value": {
          "type": "number"
        }
        "formula": {
          "type": "string"
        }
      }
    }
  }
}

Ich habe sink.go wie folgt erstellt.

`sink.go`


package sample

import (
	"fmt"
	"math"

	"gopkg.in/sensorbee/sensorbee.v0/bql"
	"gopkg.in/sensorbee/sensorbee.v0/core"
	"gopkg.in/sensorbee/sensorbee.v0/data"
)

type sink struct {
	decimal int
}

func (s *sink) Write(ctx *core.Context, tuple *core.Tuple) error {
	p, ok := tuple.Data["payload"]
	if !ok {
		return fmt.Errorf("the tuple doesn't have the required field: payload")
	}

	payload, err := data.AsMap(p)
	if err != nil {
		return err
	}

	v, ok := payload["value"]
	if !ok {
		return fmt.Errorf("the tuple doesn't have the required field: value")
	}

	value, err := data.AsFloat(v)
	if err != nil {
		return err
	}

	f, ok := payload["formula"]
	if !ok {
		return fmt.Errorf("the tuple doesn't have the required field: formula")
	}

	formula, err := data.AsString(f)
	if err != nil {
		return err
	}
	shift := math.Pow(10, float64(s.decimal))
	value = math.Floor(value*shift) / shift
	ctx.Log().Infof("formula: %s", formula)
	ctx.Log().Infof("value: %f", value)
	return nil
}

func (s *sink) Close(ctx *core.Context) error {
	return nil
}

func CreateSink(ctx *core.Context, ioParams *bql.IOParams, params data.Map) (core.Sink, error) {
	decimal, err := getDecimal(params)
	if err != nil {
		return nil, err
	}

	return &sink{
		decimal: decimal,
	}, nil
}

func getDecimal(params data.Map) (int, error) {
	node, ok := params["decimal"]
	if !ok {
		return 0, fmt.Errorf("decimal is required")
	}
	decimal, err := data.AsInt(node)
	if err != nil {
		return 0, fmt.Errorf("decimal must be a int:%s", err)
	}
	return int(decimal), nil
}

** ・ udsf.go (Verarbeitung der Eingabe / Ausgabe-Konvertierung) ** Das udsf-Objekt empfängt die folgenden Daten: --String-Typ-Parameter: stream_name (Eingabe Stream-Name) --Operator

Schwimmpunkttyp-Parameter: initial_value

Das udsf-Objekt berechnet weiterhin den Wert des empfangenen Tupelwertelements mit dem im aktuellen Wert angegebenen Operator (initial_value at start).

Ich habe udsf.go wie folgt erstellt.

`udsf.go`


package sample

import (
	"fmt"

	"gopkg.in/sensorbee/sensorbee.v0/bql/udf"
	"gopkg.in/sensorbee/sensorbee.v0/core"
	"gopkg.in/sensorbee/sensorbee.v0/data"
)

type operator byte

const (
	none    = ' '
	plus    = '+'
	minus   = '-'
	times   = '*'
	divided = '/'
)

type udsf struct {
	cur float64
	ope operator
}

func (u *udsf) Process(ctx *core.Context, tuple *core.Tuple, w core.Writer) error {
	p, ok := tuple.Data["payload"]
	if !ok {
		return fmt.Errorf("the tuple doesn't have the required field: payload")
	}

	payload, err := data.AsMap(p)
	if err != nil {
		return err
	}

	v, ok := payload["value"]
	if !ok {
		return fmt.Errorf("the tuple doesn't have the required field: value")
	}

	value, err := data.AsFloat(v)
	if err != nil {
		return err
	}

	var formula string
	newVal := u.cur
	switch u.ope {
	case plus:
		newVal += value
	case minus:
		newVal -= value
	case times:
		newVal *= value
	case divided:
		newVal /= value
	}
	formula = fmt.Sprintf("%f %s %f", u.cur, string(u.ope), value)
	ctx.Log().Debug("calculate: " + formula)
	m := data.Map{
		"value":   data.Float(newVal),
		"formula": data.String(formula),
	}
	if err := w.Write(ctx, core.NewTuple(data.Map{"payload": m})); err != nil {
		return err
	}
	u.cur = newVal
	return nil
}

func (u *udsf) Terminate(ctx *core.Context) error {
	return nil
}

func CreateUDSF(decl udf.UDSFDeclarer, params data.Map) (udf.UDSF, error) {
	inputStream, err := getStreamName(params)
	if err != nil {
		return nil, err
	}

	operator, err := getOperator(params)
	if err != nil {
		return nil, err
	}

	initialValue, err := getInitialValue(params)
	if err != nil {
		return nil, err
	}

	if err := decl.Input(inputStream, nil); err != nil {
		return nil, err
	}

	return &udsf{
		ope: operator,
		cur: initialValue,
	}, nil
}

func getStreamName(params data.Map) (string, error) {
	node, ok := params["stream_name"]
	if !ok {
		return "", fmt.Errorf("stream_name is required")
	}
	streamName, err := data.AsString(node)
	if err != nil {
		return "", fmt.Errorf("stream_name must be a string:%s", err)
	}
	return streamName, nil
}

func getOperator(params data.Map) (operator, error) {
	node, ok := params["operator"]
	if !ok {
		return none, fmt.Errorf("operator is required")
	}
	operatorStr, err := data.AsString(node)
	if err != nil {
		return none, fmt.Errorf("operator must be a string:%s", err)
	}

	switch operatorStr {
	case "plus":
		return plus, nil
	case "minus":
		return minus, nil
	case "times":
		return times, nil
	case "divided":
		return divided, nil
	default:
		return none, fmt.Errorf("invalid oparator")
	}
}

func getInitialValue(params data.Map) (float64, error) {
	initialValue := 0.0
	node, ok := params["initial_value"]
	if !ok {
		return initialValue, nil
	}
	initialValue, err := data.AsFloat(node)
	if err != nil {
		return initialValue, fmt.Errorf("initial_value is invalid")
	}
	return initialValue, nil
}

** Registrierung der Quelldatei ** Erstellen Sie plugin.go im Plugin-Verzeichnis und implementieren Sie den Registrierungsprozess für Source, Sink und UDSF Operation (zur Verwendung als BQL).

Ich habe plugin.go wie folgt erstellt.

`plugin.go`


package plugin

import (
	"github.sis.saison.co.jp/sherpa/es-agent/sample"
	"gopkg.in/sensorbee/sensorbee.v0/bql"
	"gopkg.in/sensorbee/sensorbee.v0/bql/udf"
)

func init() {
	bql.MustRegisterGlobalSourceCreator("sample_source", bql.SourceCreatorFunc(sample.CreateSource))
	bql.MustRegisterGlobalSinkCreator("sample_sink", bql.SinkCreatorFunc(sample.CreateSink))
	udf.MustRegisterGlobalUDSFCreator("sample_udsf", udf.MustConvertToUDSFCreator(sample.CreateUDSF))
}

** Erstellen der Hauptfunktion ** Erstellen und implementieren Sie schließlich plugin_main.go im externen Verzeichnis der Hauptfunktion, die den Prozess als einzelnes Ausführungsmodul aufruft.

Ich habe plugin_main.go wie folgt erstellt.

`plugin_main.go`



package main

import (
	"os"

	"github.sis.saison.co.jp/sherpa/es-agent/external/plugin"
	_ "github.sis.saison.co.jp/sherpa/es-agent/sample/plugin"
)

func main() {
	if err := plugin.NewServer().ListenAndServe(); err != nil {
		os.Exit(1)
	}
}

Zu diesem Zeitpunkt ist die Vorbereitung auf der Laufzeitseite abgeschlossen. Als nächstes werden wir die Seite der Entwicklungsumgebung implementieren.

Java-Implementierung (Development Environment Side)

Implementiert die BQL-Generierung (SQL-like Syntax), die Runtime mitteilt, wie die Streaming-Verarbeitung durchgeführt werden soll.

** Umweltvorbereitung ** ** Erstellen eines Modulverzeichnisses **

Erstellen Sie das Verzeichnis "$ SDK_HOME / dev / sample_adapter".

** Datei kopieren ** -Kopieren Sie build.xml und config.properties im Verzeichnis $ SDK_HOME / dev / conf in das Modulverzeichnis. $SDK_HOME/dev/conf/build.xml ⇒ $SDK_HOME/dev/sample_adapter/build.xml $SDK_HOME/dev/conf/config.properties ⇒ $SDK_HOME/dev/sample_adapter/config.propertites

Bearbeiten Sie die kopierte Datei config.properties.

Implementation-Title=SampleAdapter
Implementation-Vendor=sugimon
Implementation-Version=0

module.category=Sample
module.label=Sample Plugin
display.name=Sample Plugin Adapter

plugin.name=sample_plugin
esagent.plugin.package=github.sis.saison.co.jp/sherpa/es-agent/sample

** Erstellen Sie ein Verzeichnis für Quelldateien ** Erstellen Sie ein src-Verzeichnis in $ SDK_HOME / dev / sample_adapter. ($SDK_HOME/dev/sample_adapter/src)

Erstellen Sie als Nächstes das Java-Dateipaket "com / appresso / ds / dp / modules / adapter / sample" Erstellen Sie ein Verzeichnis für die folgenden Pakete, damit ($SDK_HOME/dev/sample_adapter/src/com/appresso/ds/dp/modules/adapter/sample)

** Erstellen einer Quelldatei ** Erstellen Sie eine Quelldatei im Paketverzeichnis mit dem folgenden Dateinamen.

・SampleAdapterModuleComponent.java ・SampleSinkOperationFactory.java ・SampleSourceOperationFactory.java ・SampleUDSFOperationFactory.java

├─ sample_adapter
│      │  build.xml
│      │  config.properties
│      ├─ src
│      │   └com
│      │     └appresso
│      │       └ds
│      │         └dp
│      │           └modules
│      │              └adapter
│      │                 └sample
│      │                    SampleAdapterModuleComponent.java
│      │                    SampleSinkOperationFactory.java
│      │                    SampleSourceOperationFactory.java
│      │                    SampleUDSFOperationFactory.java

Lassen Sie uns auch für jede dieser Dateien eine Quelldatei erstellen.

** ・ SampleSourceOperationFactory.java (Eingabeverarbeitung) ** Gibt ein Objekt zurück, das die Eigenschaften der Quelloperation enthält, oder ein Operationsobjekt. (Vererbungsquellenklasse: BQLPluginSourceOperationFactory-Klasse)

Ich habe SampleSourceOperationFactory.java wie folgt erstellt.

`SampleSourceOperationFactory.java`


package com.appresso.ds.dp.modules.adapter.sample;

import com.appresso.ds.common.spi.constraint.NumberFillin;
import com.appresso.ds.common.spi.param.SimpleParameter;
import com.appresso.ds.common.xmlfw.xml.XmlHandler;
import com.appresso.ds.dp.share.adapter.bql.common.BQLPluginSourceOperationFactory;
import com.appresso.ds.dp.spi.OperationConfiguration;
import com.appresso.ds.dp.spi.OperationConfigurator;
import org.xml.sax.SAXException;

import static com.appresso.ds.common.bql.BQLSimpleParameterType.FLOAT;

public class SampleSourceOperationFactory extends BQLPluginSourceOperationFactory {

	@Override
	protected String getLabel() {
		return "Sample source";
	}

	@Override
	protected String getPluginName() {
		return "sample_plugin";
	}

	@Override
	public String getOperationName() {
		return "sample_source";
	}

	@Override
	protected String getTypeName() {
		return "sample_source";
	}

	@Override
	protected void setupOperationConfigurator(OperationConfigurator operationConfigurator) {
		operationConfigurator.addSimpleParameter(createIntervalParameter());
	}

	@Override
	protected void setupOutputSchema(XmlHandler handler, OperationConfiguration conf) throws Exception {
		handler.startElement("", "payload", "payload", EMPTY_ATTRIBUTE);
		writeElement(handler, "value");
		handler.endElement("", "payload", "payload");
	}

	protected void writeElement(XmlHandler handler, String name) throws SAXException {
		handler.startElement("", name, name, EMPTY_ATTRIBUTE);
		handler.endElement("", name, name);
	}

	static SimpleParameter createIntervalParameter() {
		NumberFillin fillin = new NumberFillin();
		fillin.setMinValue(0.001);
		fillin.setMaxValue(1314000);
		fillin.setAllowMin(true);
		fillin.setAllowMax(true);
		fillin.setPrecision(10);
		fillin.setDecimal(3);
		fillin.setAllowDouble(true);
		fillin.setLabel("Interval[sec]");
		fillin.setRequired(true);
		return new SimpleParameter(FLOAT.toParameterKey("interval"), fillin);
	}
}

** ・ SampleSinkOperationFactory.java (Ausgabeverarbeitung) ** Gibt ein Objekt zurück, das die Eigenschaften einer Sink-Operation oder eines Operationsobjekts enthält. (Vererbungsquellenklasse: BQLPluginSinkOperationFactory-Klasse)

Ich habe SampleSinkOperationFactory.java wie folgt erstellt.

`SampleSinkOperationFactory.java`


package com.appresso.ds.dp.modules.adapter.sample;

import com.appresso.ds.common.spi.constraint.NumberFillin;
import com.appresso.ds.common.spi.param.SimpleParameter;
import com.appresso.ds.common.xmlfw.xml.XmlHandler;
import com.appresso.ds.dp.share.adapter.bql.common.BQLPluginOutputOperationFactory;
import com.appresso.ds.dp.spi.OperationConfiguration;
import com.appresso.ds.dp.spi.OperationConfigurator;
import com.appresso.ds.dp.spi.OperationContext;
import org.xml.sax.SAXException;

import static com.appresso.ds.common.bql.BQLSimpleParameterType.INTEGER;

public class SampleSinkOperationFactory extends BQLPluginOutputOperationFactory {
	@Override
	protected String getLabel() {
		return "Sample sink";
	}

	@Override
	protected String getPluginName() {
		return "sample_plugin";
	}

	@Override
	public String getOperationName() {
		return "sample_sink";
	}

	@Override
	protected String getTypeName() {
		return "sample_sink";
	}

	@Override
	protected void setupOperationConfigurator(OperationConfigurator operationConfigurator) {
		operationConfigurator.addSimpleParameter(createDecimalParameter());
	}

	protected void setupInputSchema(XmlHandler handler, OperationConfiguration conf, OperationContext context)
			throws Exception {
		handler.startElement("", "payload", "payload", EMPTY_ATTRIBUTE);
		writeElement(handler, "formula");
		writeElement(handler, "value");
		handler.endElement("", "payload", "payload");
	}

	protected void writeElement(XmlHandler handler, String name) throws SAXException {
		handler.startElement("", name, name, EMPTY_ATTRIBUTE);
		handler.endElement("", name, name);
	}

	static SimpleParameter createDecimalParameter() {
		NumberFillin fillin = new NumberFillin();
		fillin.setMinValue(0);
		fillin.setMaxValue(10);
		fillin.setAllowMin(true);
		fillin.setAllowMax(true);
		fillin.setLabel("Decimal");
		fillin.setRequired(true);
		return new SimpleParameter(INTEGER.toParameterKey("decimal"), fillin);
	}
}

** ・ SampleUDSFOperationFactory ** Gibt ein Objekt zurück, das die Eigenschaften einer UDSF-Operation oder eines Operationsobjekts enthält.

Ich habe SampleUDSFOperationFactory.java wie folgt erstellt.

`SampleUDSFOperationFactory.java`


package com.appresso.ds.dp.modules.adapter.sample;

import com.appresso.ds.common.bql.UDSFFromArgument;
import com.appresso.ds.common.bql.UDSFFromTemplate;
import com.appresso.ds.common.spi.constraint.Item;
import com.appresso.ds.common.spi.constraint.Multi;
import com.appresso.ds.common.spi.constraint.NumberFillin;
import com.appresso.ds.common.spi.param.SimpleParameter;
import com.appresso.ds.common.xmlfw.xml.XmlHandler;
import com.appresso.ds.dp.share.adapter.bql.common.BQLPluginUDSFOperationFactory;
import com.appresso.ds.dp.spi.OperationConfiguration;
import com.appresso.ds.dp.spi.OperationConfigurator;
import com.appresso.ds.dp.spi.OperationContext;
import org.xml.sax.SAXException;

import java.util.stream.Stream;

import static com.appresso.ds.common.bql.BQLSimpleParameterType.FLOAT;
import static com.appresso.ds.common.bql.BQLSimpleParameterType.STRING;

public class SampleUDSFOperationFactory extends BQLPluginUDSFOperationFactory {
	@Override
	protected String getLabel() {
		return "Sample UDSF";
	}

	@Override
	public String getPluginName() {
		return "sample_plugin";
	}

	@Override
	protected String getTypeName() {
		return "sample_udsf";
	}

	@Override
	public String getOperationName() {
		return "sample_udsf";
	}

	@Override
	protected void addArgs(UDSFFromTemplate template) {
		template.addArg(new UDSFFromArgument(STRING.toParameterKey("operator")));
		template.addArg(new UDSFFromArgument(FLOAT.toParameterKey("initial_value")));
	}

	@Override
	protected void setupOperationConfigurator(OperationConfigurator operationConfigurator) {
		setStreamConfigurationParameter(operationConfigurator);
		operationConfigurator.addSimpleParameter(createOperatorParameter());
		operationConfigurator.addSimpleParameter(createInitialValueParameter());
	}

	@Override
	protected void setupInputSchema(XmlHandler handler, OperationConfiguration conf, OperationContext context)
			throws Exception {
		handler.startElement("", "payload", "payload", EMPTY_ATTRIBUTE);
		writeElement(handler, "value");
		handler.endElement("", "payload", "payload");
	}

	@Override
	protected void setupOutputSchema(XmlHandler handler, OperationConfiguration conf, OperationContext context)
			throws Exception {
		handler.startElement("", "payload", "payload", EMPTY_ATTRIBUTE);
		writeElement(handler, "formula");
		writeElement(handler, "value");
		handler.endElement("", "payload", "payload");
	}

	protected void writeElement(XmlHandler handler, String name) throws SAXException {
		handler.startElement("", name, name, EMPTY_ATTRIBUTE);
		handler.endElement("", name, name);
	}

	static SimpleParameter createInitialValueParameter() {
		NumberFillin fillin = new NumberFillin();
		fillin.setPrecision(10);
		fillin.setDecimal(3);
		fillin.setAllowDouble(true);
		fillin.setLabel("Initial value");
		fillin.setRequired(true);
		return new SimpleParameter(FLOAT.toParameterKey("initial_value"), fillin);
	}

	static SimpleParameter createOperatorParameter(){
		Multi multi = new Multi(Operator.getItems());
		multi.setLabel("Operator");
		multi.setRequired(true);
		SimpleParameter param = new SimpleParameter(STRING.toParameterKey("operator"), multi);
		return param;
	}

	enum Operator {
		Plus("+","plus"),
		Minus("-","minus"),
		Times("*","times"),
		Divided("/","divided");

		public String getDisplayName() {
			return displayName;
		}

		public String getValue() {
			return value;
		}

		private final String displayName;

		private final String value;

		private Operator(String displayName, String value) {
			this.displayName = displayName;
			this.value=value;
		}

		Item toItem(){
			return new Item(value,displayName);
		}

		static Item[] getItems(){
			return Stream.of(Operator.values()).map(s->s.toItem()).toArray(Item[]::new);
		}
	}
}

** ・ SampleAdapterModuleComponent ** Eine Klasse, die eine Komponente (Adapter) darstellt.

Ich habe SampleAdapterModuleComponent.java wie folgt erstellt.

`SampleAdapterModuleComponent.java`


package com.appresso.ds.dp.modules.adapter.sample;

import java.util.ArrayList;
import java.util.List;

import com.appresso.ds.common.kernel.modules.LicenseManager;
import com.appresso.ds.common.license.LicensePackageType;
import com.appresso.ds.dp.spi.AdapterModuleComponent;
import com.appresso.ds.dp.spi.OperationFactory;
import com.appresso.ds.dp.spi.ResourceFactory;

public class SampleAdapterModuleComponent extends AdapterModuleComponent {

	private static final String MODULE_COMPONENT_NAME = "Sample Adapter";

	@Override
	public OperationFactory[] getOperationFactories() throws Exception {
		List<OperationFactory> operationFactories = new ArrayList<>();
		operationFactories.add(new SampleSourceOperationFactory());
		operationFactories.add(new SampleUDSFOperationFactory());
		operationFactories.add(new SampleSinkOperationFactory());
		return operationFactories.toArray(new OperationFactory[operationFactories.size()]);
	}

	@Override
	public ResourceFactory[] getResourceFactories() throws Exception {
		return new ResourceFactory[]{};
	}

	public void checkLicense() throws Exception {
		LicenseManager licenseManager = getContext().getProxy(LicenseManager.class);
		licenseManager.checkLicense(getModuleComponentName(), getPermittedPackageType());
	}

	private String getModuleComponentName() {
		return MODULE_COMPONENT_NAME;
	}

	private int[] getPermittedPackageType() {
		return new int[]{LicensePackageType.TYPE_BASIC_SERVER};
	}
}

Schließlich

Dieses Mal habe ich die Verarbeitung tatsächlich getrennt für die Seite der Ausführungsumgebung und die Seite der Entwicklungsumgebung implementiert. Damit ist der Plug-In-Erstellungsprozess abgeschlossen. Nächstes Mal möchte ich diese erstellen und ausführen.

In diesem Blog möchte ich weiterhin die Inhalte der Konsultationen am "Yorozu Consultation Counter" der Technologie und die geborenen Tricks vorstellen.

Bitte probieren Sie es weiter aus und folgen Sie uns, wenn Sie möchten.

Wir sehen uns wieder!

[GO] Ich habe versucht, ein Plug-In mit HULFT IoT Edge Streaming [Entwicklung] (2/3) zu erstellen.

Einführung

Was ist diesmal zu tun?

Übersicht über das Erstellen eines Plug-Ins

Konfiguration der Ausführungsumgebung

Konfiguration der Entwicklungsumgebung

Plug-In erstellen

Implementierung auf der Seite der Ausführungsumgebung (Golang)

Tuple

source.go

Tuple

sink.go

udsf.go

plugin.go

plugin_main.go

Java-Implementierung (Development Environment Side)

SampleSourceOperationFactory.java

SampleSinkOperationFactory.java

SampleUDSFOperationFactory.java

SampleAdapterModuleComponent.java

Schließlich

`Tuple`

`source.go`

`Tuple`

`sink.go`

`udsf.go`

`plugin.go`

`plugin_main.go`

`SampleSourceOperationFactory.java`

`SampleSinkOperationFactory.java`

`SampleUDSFOperationFactory.java`

`SampleAdapterModuleComponent.java`