// package client_bus_http -- клиент HTTP-шины. package client_bus_http import ( "encoding/json" "fmt" "io" "net/http" "strings" "github.com/google/uuid" mL0 "gitp78su.ipnodns.ru/svi/kern/v4/lev0" "gitp78su.ipnodns.ru/svi/kern/v4/lev0/defs/topic" mKs "gitp78su.ipnodns.ru/svi/kern/v4/lev0/kspec" mL1 "gitp78su.ipnodns.ru/svi/kern/v4/lev1" "gitp78su.ipnodns.ru/svi/kern/v4/lev1/kbus_msg/msg_pub" "gitp78su.ipnodns.ru/svi/kern/v4/lev1/kbus_msg/msg_serve" "gitp78su.ipnodns.ru/svi/kern/v4/lev1/kbus_msg/msg_sub" "gitp78su.ipnodns.ru/svi/kern/v4/lev1/kbus_msg/msg_unsub" "gitp78su.ipnodns.ru/svi/kern/v4/lev1/log_buf" "gitp78su.ipnodns.ru/svi/kern/v4/lev1/safe_bool" "gitp78su.ipnodns.ru/svi/kern/v4/lev1/safe_string" "gitp78su.ipnodns.ru/svi/kern/v4/lev2/kbus_http" "gitp78su.ipnodns.ru/svi/kern/v4/lev2/kctx" ) // ClientBusHttp -- клиент HTTP-шины. type ClientBusHttp struct { bus mKs.IKernelBus lCtx mKs.ILocalCtx log mKs.ILogBuf isWork mKs.ISafeBool urlRemote string // URL дистанционной шины urlLocal string // URL локальной шины для веб-хуков } // NewClientBusHttp - -возвращает новый клиент HTTP-шины. func NewClientBusHttp(urlRemote string) mKs.IBusClient { param := &log_buf.LogBufParam{ IsTerm_: safe_bool.NewSafeBool(true), Prefix_: "ClientBusHttp", } log := mL1.NewLogBuf(param) log.Debug("NewClientBusHttp()") mL0.Hassert(urlRemote != "", "NewClientBusHttp(): urlRemote is empty") kCtx := kctx.GetKernelCtx() resLocal := safe_string.NewSafeStringGetenv("LOCAL_HTTP_URL") resLocal.Hassert("NewClientBusHttp(): in get env LOCAL_HTTP_URL)") urlLocal := resLocal.Ok() kBusHttp := kbus_http.GetKernelBusHttp() sf := &ClientBusHttp{ lCtx: mL1.NewLocalCtx(kCtx.Ctx()), log: log, bus: kBusHttp, isWork: mL1.NewSafeBool(), urlRemote: strings.TrimSuffix(urlRemote, "/"), urlLocal: strings.TrimSuffix(urlLocal.Get(), "/"), } return sf } // Unsubscribe -- отписывается от топика в дистанционной шине. func (sf *ClientBusHttp) Unsubscribe(handler mKs.IBusHandlerSubscribe) { _uuid, err := uuid.NewV6() mL1.Hassert()(err == nil, "ClientBusHttp.Unsubscribe(): in generate UUID v6, err=\n\t%v", err) req := &msg_unsub.UnsubReq{ Name_: handler.Name(), Uuid_: _uuid.String(), } req.SelfCheck() binReq, err := json.MarshalIndent(req, "", " ") mL1.Hassert()(err == nil, "ClientBusHttp.Unsubscribe(): in marshal to JSON unsubscribe request, err=\n\t%v", err) reader := strings.NewReader(string(binReq)) hReq, err := http.NewRequest(http.MethodPost, sf.urlRemote+"/bus/unsub", reader) mL1.Hassert()(err == nil, "ClientBusHttp.Unsubscribe(): in new request, err=\n\t%v", err) resBody := sf.makePost(hReq) if resBody.IsErr() { return } resp := &msg_unsub.UnsubResp{} err = json.Unmarshal(resBody.Ok(), resp) if err != nil { sf.log.Err("Unsubscribe(): in unmarshal response, err=\n\t%v", err) return } if resp.Status_ != "ok" { sf.log.Err("Unsubscribe(): resp!='ok', err=\n\t%v", resp.Status_) } if resp.Uuid_ != req.Uuid_ { sf.log.Err("Unsubscribe(): resp uuid(%v) bad", resp.Uuid_) } } // Subscribe -- подписывается на топик в дистанционной шине. func (sf *ClientBusHttp) Subscribe(handler mKs.IBusHandlerSubscribe) mKs.IResult[bool] { _uuid, err := uuid.NewV6() mL0.Hassert(err == nil, "ClientBusHttp.Subscribe(): in generate UUID v6, err=\n\t%v", err) req := &msg_sub.SubscribeReq{ Topic_: handler.Topic(), Uuid_: _uuid.String(), WebHook_: sf.urlLocal + "/bus/pub", // Веб-хук всегда на своей стороне } req.SelfCheck() binReq, _ := json.MarshalIndent(req, "", " ") body := strings.NewReader(string(binReq)) hReq, err := http.NewRequest(http.MethodPost, sf.urlRemote+"/bus/sub", body) mL0.Hassert(err != nil, "ClientBusHttp.Subscribe(): in new request, err=\n\t%v") resBody := sf.makePost(hReq) if resBody.IsErr() { err := fmt.Errorf("ClientBusHttp.Subscribe(): in make POST-request, err=\n\t%w", err) return mL0.NewErr[bool](err) } resp := &msg_sub.SubscribeResp{} err = json.Unmarshal(resBody.Ok(), resp) if err != nil { err := fmt.Errorf("ClientBusHttp.Subscribe(): in unmarshal response, err=\n\t%v", err) return mL0.NewErr[bool](err) } if resp.Status_ != "ok" { err := fmt.Errorf("ClientBusHttp.Subscribe(): resp!='ok', err=\n\t%v", resp.Status_) return mL0.NewErr[bool](err) } if resp.Uuid_ != req.Uuid_ { err := fmt.Errorf("ClientBusHttp.Subscribe(): resp uuid(%v) bad", resp.Uuid_) return mL0.NewErr[bool](err) } res := sf.bus.Subscribe(handler) return res } // SendRequest -- отправляет в дистанционную шину запрос. func (sf *ClientBusHttp) SendRequest(topic *topic.ATopic, binReq []byte) mL0.IResult[[]byte] { _uuid, err := uuid.NewV6() mL1.Hassert()(err == nil, "ClientBusHttp.SendRequest(): in generate UUID v6, err=\n\t%v", err) req := &msg_serve.ServeReq{ Topic_: topic, Uuid_: _uuid.String(), BinReq_: binReq, } req.SelfCheck() _binReq, _ := json.MarshalIndent(req, "", " ") body := strings.NewReader(string(_binReq)) hReq, err := http.NewRequest(http.MethodPost, sf.urlRemote+"/bus/request", body) mL1.Hassert()(err == nil, "ClientBusHttp.SendRequest(): in new request, err=\n\t%v") resBody := sf.makePost(hReq) if resBody.IsErr() { err := fmt.Errorf("ClientBusHttp.SendRequest(): in make request, err=\n\t%w", err) return mL0.WrapErr(resBody, err) } resp := &msg_serve.ServeResp{} err = json.Unmarshal(resBody.Ok(), resp) if err != nil { err := fmt.Errorf("ClientBusHttp.SendRequest(): in unmarshal response, err=\n\t%v", err) return mL0.NewErr[[]byte](err) } if resp.Status_ != "ok" { err := fmt.Errorf("ClientBusHttp.SendRequest(): resp!='ok', err=\n\t%v", resp.Status_) return mL0.NewErr[[]byte](err) } if resp.Uuid_ != req.Uuid_ { err := fmt.Errorf("ClientBusHttp.SendRequest(): resp uuid(%v) bad", resp.Uuid_) return mL0.NewErr[[]byte](err) } return mL0.NewOk(resp.BinResp_) } // RegisterServe -- регистрирует в локальной шине обработчик. func (sf *ClientBusHttp) RegisterServe(handler mKs.IBusHandlerServe) mL0.IResult[bool] { if handler == nil { return mL0.NewErr[bool](fmt.Errorf("ClientBusHttp.RegisterServe(): handler==nil")) } res := sf.bus.RegisterServe(handler) if res.IsErr() { err := fmt.Errorf("ClientBusHttp.RegisterServe(): in register serve to bus, err=\n\t%v", res.Err()) return mL0.NewErr[bool](err) } return mL0.NewOk(true) } // Publish -- публикует сообщение в дистанционной шину. func (sf *ClientBusHttp) Publish(topic *topic.ATopic, binMsg []byte) mL0.IResult[bool] { _uuid, err := uuid.NewV6() mL1.Hassert()(err == nil, "ClientBusHttp.Publish(): in generate UUID v6, err=\n\t%v", err) req := &msg_pub.PublishReq{ Topic_: topic, Uuid_: _uuid.String(), BinMsg_: binMsg, } req.SelfCheck() binReq, _ := json.MarshalIndent(req, "", " ") body := strings.NewReader(string(binReq)) hReq, err := http.NewRequest(http.MethodPost, sf.urlRemote+"/bus/pub", body) mL1.Hassert()(err == nil, "ClientBusHttp.Publish(): in new request, err=\n\t%v") resBody := sf.makePost(hReq) if resBody.IsErr() { err := fmt.Errorf("ClientBusHttp.Publish(): in make request, err=\n\t%w", err) return mL0.NewErr[bool](err) } resp := &msg_pub.PublishResp{} err = json.Unmarshal(resBody.Ok(), resp) if err != nil { err := fmt.Errorf("ClientBusHttp.Publish(): in unmarshal response, err=\n\t%v", err) return mL0.NewErr[bool](err) } if resp.Status_ != "ok" { err := fmt.Errorf("ClientBusHttp.Publish(): resp!='ok', err=\n\t%v", resp.Status_) return mL0.NewErr[bool](err) } if resp.Uuid_ != req.Uuid_ { err := fmt.Errorf("ClientBusHttp.Publish(): resp uuid(%v) bad", resp.Uuid_) return mL0.NewErr[bool](err) } return mL0.NewOk(true) } // Единый обработчик запросов. func (sf *ClientBusHttp) makePost(hReq *http.Request) mKs.IResult[[]byte] { hReq.Header.Add("Content-Type", "application/json") _resp, err := http.DefaultClient.Do(hReq) if err != nil { err := fmt.Errorf("ClientBusHttp.makePost(): after request, err=\n\t%w", err) return mL0.NewErr[[]byte](err) } defer func() { _ = _resp.Body.Close() }() if _resp.StatusCode != http.StatusOK { err := fmt.Errorf("ClientBusHttp.makePost(): url=%q, status=%q", hReq.URL, _resp.Status) return mL0.NewErr[[]byte](err) } binBody, err := io.ReadAll(_resp.Body) if err != nil { err := fmt.Errorf("ClientBusHttp.makePost(): after read body, err=\n\t%v", err) return mL0.NewErr[[]byte](err) } return mL0.NewOk(binBody) } // Log -- возвращает локальный лог клиента. func (sf *ClientBusHttp) Log() mKs.ILogBuf { return sf.log } // IsWork -- возвращает признак работы. func (sf *ClientBusHttp) IsWork() mKs.EBool { return sf.bus.IsWork() }