api.go 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595
  1. package api
  2. import (
  3. ctx "context"
  4. "connectrpc.com/connect"
  5. apiv1 "github.com/OliveTin/OliveTin/gen/olivetin/api/v1"
  6. apiv1connect "github.com/OliveTin/OliveTin/gen/olivetin/api/v1/apiv1connect"
  7. "github.com/google/uuid"
  8. log "github.com/sirupsen/logrus"
  9. "fmt"
  10. "net/http"
  11. acl "github.com/OliveTin/OliveTin/internal/acl"
  12. config "github.com/OliveTin/OliveTin/internal/config"
  13. executor "github.com/OliveTin/OliveTin/internal/executor"
  14. installationinfo "github.com/OliveTin/OliveTin/internal/installationinfo"
  15. sv "github.com/OliveTin/OliveTin/internal/stringvariables"
  16. )
  17. type oliveTinAPI struct {
  18. executor *executor.Executor
  19. cfg *config.Config
  20. connectedClients []*connectedClients
  21. }
  22. type connectedClients struct {
  23. channel chan *apiv1.EventStreamResponse
  24. AuthenticatedUser *acl.AuthenticatedUser
  25. }
  26. func (api *oliveTinAPI) KillAction(ctx ctx.Context, req *connect.Request[apiv1.KillActionRequest]) (*connect.Response[apiv1.KillActionResponse], error) {
  27. ret := &apiv1.KillActionResponse{
  28. ExecutionTrackingId: req.Msg.ExecutionTrackingId,
  29. }
  30. var execReqLogEntry *executor.InternalLogEntry
  31. execReqLogEntry, ret.Found = api.executor.GetLog(req.Msg.ExecutionTrackingId)
  32. if !ret.Found {
  33. log.Warnf("Killing execution request not possible - not found by tracking ID: %v", req.Msg.ExecutionTrackingId)
  34. return connect.NewResponse(ret), nil
  35. }
  36. log.Warnf("Killing execution request by tracking ID: %v", req.Msg.ExecutionTrackingId)
  37. action := api.cfg.FindAction(execReqLogEntry.ActionTitle)
  38. if action == nil {
  39. log.Warnf("Killing execution request not possible - action not found: %v", execReqLogEntry.ActionTitle)
  40. ret.Killed = false
  41. return connect.NewResponse(ret), nil
  42. }
  43. user := acl.UserFromContext(ctx, api.cfg)
  44. api.killActionByTrackingId(user, action, execReqLogEntry, ret)
  45. return connect.NewResponse(ret), nil
  46. }
  47. func (api *oliveTinAPI) killActionByTrackingId(user *acl.AuthenticatedUser, action *config.Action, execReqLogEntry *executor.InternalLogEntry, ret *apiv1.KillActionResponse) {
  48. if !acl.IsAllowedKill(api.cfg, user, action) {
  49. log.Warnf("Killing execution request not possible - user not allowed to kill this action: %v", execReqLogEntry.ExecutionTrackingID)
  50. ret.Killed = false
  51. }
  52. err := api.executor.Kill(execReqLogEntry)
  53. if err != nil {
  54. log.Warnf("Killing execution request err: %v", err)
  55. ret.AlreadyCompleted = true
  56. ret.Killed = false
  57. } else {
  58. ret.Killed = true
  59. }
  60. }
  61. func (api *oliveTinAPI) StartAction(ctx ctx.Context, req *connect.Request[apiv1.StartActionRequest]) (*connect.Response[apiv1.StartActionResponse], error) {
  62. args := make(map[string]string)
  63. for _, arg := range req.Msg.Arguments {
  64. args[arg.Name] = arg.Value
  65. }
  66. api.executor.MapActionIdToBindingLock.RLock()
  67. pair := api.executor.MapActionIdToBinding[req.Msg.ActionId]
  68. api.executor.MapActionIdToBindingLock.RUnlock()
  69. if pair == nil || pair.Action == nil {
  70. return nil, connect.NewError(connect.CodeNotFound, fmt.Errorf("action with ID %s not found", req.Msg.ActionId))
  71. }
  72. authenticatedUser := acl.UserFromContext(ctx, api.cfg)
  73. execReq := executor.ExecutionRequest{
  74. Action: pair.Action,
  75. EntityPrefix: pair.EntityPrefix,
  76. TrackingID: req.Msg.UniqueTrackingId,
  77. Arguments: args,
  78. AuthenticatedUser: authenticatedUser,
  79. Cfg: api.cfg,
  80. }
  81. api.executor.ExecRequest(&execReq)
  82. ret := &apiv1.StartActionResponse{
  83. ExecutionTrackingId: execReq.TrackingID,
  84. }
  85. return connect.NewResponse(ret), nil
  86. }
  87. func (api *oliveTinAPI) PasswordHash(ctx ctx.Context, req *connect.Request[apiv1.PasswordHashRequest]) (*connect.Response[apiv1.PasswordHashResponse], error) {
  88. hash, err := createHash(req.Msg.Password)
  89. if err != nil {
  90. return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("error creating hash: %w", err))
  91. }
  92. ret := &apiv1.PasswordHashResponse{
  93. Hash: hash,
  94. }
  95. return connect.NewResponse(ret), nil
  96. }
  97. func (api *oliveTinAPI) LocalUserLogin(ctx ctx.Context, req *connect.Request[apiv1.LocalUserLoginRequest]) (*connect.Response[apiv1.LocalUserLoginResponse], error) {
  98. match := checkUserPassword(api.cfg, req.Msg.Username, req.Msg.Password)
  99. if match {
  100. //grpc.SendHeader(ctx, metadata.Pairs("set-username", req.Username))
  101. log.WithFields(log.Fields{
  102. "username": req.Msg.Username,
  103. }).Info("LocalUserLogin: User logged in successfully.")
  104. } else {
  105. log.WithFields(log.Fields{
  106. "username": req.Msg.Username,
  107. }).Warn("LocalUserLogin: User login failed.")
  108. }
  109. return connect.NewResponse(&apiv1.LocalUserLoginResponse{
  110. Success: match,
  111. }), nil
  112. }
  113. func (api *oliveTinAPI) StartActionAndWait(ctx ctx.Context, req *connect.Request[apiv1.StartActionAndWaitRequest]) (*connect.Response[apiv1.StartActionAndWaitResponse], error) {
  114. args := make(map[string]string)
  115. for _, arg := range req.Msg.Arguments {
  116. args[arg.Name] = arg.Value
  117. }
  118. user := acl.UserFromContext(ctx, api.cfg)
  119. execReq := executor.ExecutionRequest{
  120. Action: api.executor.FindActionBindingByID(req.Msg.ActionId),
  121. TrackingID: uuid.NewString(),
  122. Arguments: args,
  123. AuthenticatedUser: user,
  124. Cfg: api.cfg,
  125. }
  126. wg, _ := api.executor.ExecRequest(&execReq)
  127. wg.Wait()
  128. internalLogEntry, ok := api.executor.GetLog(execReq.TrackingID)
  129. if ok {
  130. return connect.NewResponse(&apiv1.StartActionAndWaitResponse{
  131. LogEntry: api.internalLogEntryToPb(internalLogEntry, user),
  132. }), nil
  133. } else {
  134. return nil, fmt.Errorf("execution not found")
  135. }
  136. }
  137. func (api *oliveTinAPI) StartActionByGet(ctx ctx.Context, req *connect.Request[apiv1.StartActionByGetRequest]) (*connect.Response[apiv1.StartActionByGetResponse], error) {
  138. args := make(map[string]string)
  139. execReq := executor.ExecutionRequest{
  140. Action: api.executor.FindActionBindingByID(req.Msg.ActionId),
  141. TrackingID: uuid.NewString(),
  142. Arguments: args,
  143. AuthenticatedUser: acl.UserFromContext(ctx, api.cfg),
  144. Cfg: api.cfg,
  145. }
  146. _, uniqueTrackingId := api.executor.ExecRequest(&execReq)
  147. return connect.NewResponse(&apiv1.StartActionByGetResponse{
  148. ExecutionTrackingId: uniqueTrackingId,
  149. }), nil
  150. }
  151. func (api *oliveTinAPI) StartActionByGetAndWait(ctx ctx.Context, req *connect.Request[apiv1.StartActionByGetAndWaitRequest]) (*connect.Response[apiv1.StartActionByGetAndWaitResponse], error) {
  152. args := make(map[string]string)
  153. user := acl.UserFromContext(ctx, api.cfg)
  154. execReq := executor.ExecutionRequest{
  155. Action: api.executor.FindActionBindingByID(req.Msg.ActionId),
  156. TrackingID: uuid.NewString(),
  157. Arguments: args,
  158. AuthenticatedUser: user,
  159. Cfg: api.cfg,
  160. }
  161. wg, _ := api.executor.ExecRequest(&execReq)
  162. wg.Wait()
  163. internalLogEntry, ok := api.executor.GetLog(execReq.TrackingID)
  164. if ok {
  165. return connect.NewResponse(&apiv1.StartActionByGetAndWaitResponse{
  166. LogEntry: api.internalLogEntryToPb(internalLogEntry, user),
  167. }), nil
  168. } else {
  169. return nil, connect.NewError(connect.CodeNotFound, fmt.Errorf("execution not found"))
  170. }
  171. }
  172. func (api *oliveTinAPI) internalLogEntryToPb(logEntry *executor.InternalLogEntry, authenticatedUser *acl.AuthenticatedUser) *apiv1.LogEntry {
  173. pble := &apiv1.LogEntry{
  174. ActionTitle: logEntry.ActionTitle,
  175. ActionIcon: logEntry.ActionIcon,
  176. ActionId: logEntry.ActionId,
  177. DatetimeStarted: logEntry.DatetimeStarted.Format("2006-01-02 15:04:05"),
  178. DatetimeFinished: logEntry.DatetimeFinished.Format("2006-01-02 15:04:05"),
  179. DatetimeIndex: logEntry.Index,
  180. Output: logEntry.Output,
  181. TimedOut: logEntry.TimedOut,
  182. Blocked: logEntry.Blocked,
  183. ExitCode: logEntry.ExitCode,
  184. Tags: logEntry.Tags,
  185. ExecutionTrackingId: logEntry.ExecutionTrackingID,
  186. ExecutionStarted: logEntry.ExecutionStarted,
  187. ExecutionFinished: logEntry.ExecutionFinished,
  188. User: logEntry.Username,
  189. }
  190. if !pble.ExecutionFinished {
  191. pble.CanKill = acl.IsAllowedKill(api.cfg, authenticatedUser, api.cfg.FindAction(logEntry.ActionConfigTitle))
  192. }
  193. return pble
  194. }
  195. func getExecutionStatusByTrackingID(api *oliveTinAPI, executionTrackingId string) *executor.InternalLogEntry {
  196. logEntry, ok := api.executor.GetLog(executionTrackingId)
  197. if !ok {
  198. return nil
  199. }
  200. return logEntry
  201. }
  202. func getMostRecentExecutionStatusById(api *oliveTinAPI, actionId string) *executor.InternalLogEntry {
  203. var ile *executor.InternalLogEntry
  204. logs := api.executor.GetLogsByActionId(actionId)
  205. if len(logs) == 0 {
  206. return nil
  207. } else {
  208. // Get last log entry
  209. ile = logs[len(logs)-1]
  210. }
  211. return ile
  212. }
  213. func (api *oliveTinAPI) ExecutionStatus(ctx ctx.Context, req *connect.Request[apiv1.ExecutionStatusRequest]) (*connect.Response[apiv1.ExecutionStatusResponse], error) {
  214. res := &apiv1.ExecutionStatusResponse{}
  215. user := acl.UserFromContext(ctx, api.cfg)
  216. var ile *executor.InternalLogEntry
  217. if req.Msg.ExecutionTrackingId != "" {
  218. ile = getExecutionStatusByTrackingID(api, req.Msg.ExecutionTrackingId)
  219. } else {
  220. ile = getMostRecentExecutionStatusById(api, req.Msg.ActionId)
  221. }
  222. if ile == nil {
  223. return nil, connect.NewError(connect.CodeNotFound, fmt.Errorf("execution not found for tracking ID %s or action ID %s", req.Msg.ExecutionTrackingId, req.Msg.ActionId))
  224. } else {
  225. res.LogEntry = api.internalLogEntryToPb(ile, user)
  226. }
  227. return connect.NewResponse(res), nil
  228. }
  229. /**
  230. func (api *oliveTinAPI) WatchExecution(req *apiv1.WatchExecutionRequest, srv apiv1.OliveTinApi_WatchExecutionServer) error {
  231. log.Infof("Watch")
  232. if logEntry, ok := api.executor.Logs[req.ExecutionUuid]; !ok {
  233. log.Errorf("Execution not found: %v", req.ExecutionUuid)
  234. return nil
  235. } else {
  236. if logEntry.ExecutionStarted {
  237. for !logEntry.ExecutionCompleted {
  238. tmp := make([]byte, 256)
  239. red, err := io.ReadAtLeast(logEntry.StdoutBuffer, tmp, 1)
  240. log.Infof("%v %v", red, err)
  241. srv.Send(&apiv1.WatchExecutionUpdate{
  242. Update: string(tmp),
  243. })
  244. }
  245. }
  246. return nil
  247. }
  248. }
  249. */
  250. func (api *oliveTinAPI) Logout(ctx ctx.Context, req *connect.Request[apiv1.LogoutRequest]) (*connect.Response[apiv1.LogoutResponse], error) {
  251. //user := acl.UserFromContext(ctx, cfg)
  252. //grpc.SendHeader(ctx, metadata.Pairs("logout-provider", user.Provider))
  253. //grpc.SendHeader(ctx, metadata.Pairs("logout-sid", user.SID))
  254. return nil, nil
  255. }
  256. func (api *oliveTinAPI) GetDashboardComponents(ctx ctx.Context, req *connect.Request[apiv1.GetDashboardComponentsRequest]) (*connect.Response[apiv1.GetDashboardComponentsResponse], error) {
  257. user := acl.UserFromContext(ctx, api.cfg)
  258. if user.IsGuest() && api.cfg.AuthRequireGuestsToLogin {
  259. return nil, connect.NewError(connect.CodePermissionDenied, fmt.Errorf("guests are not allowed to access the dashboard"))
  260. }
  261. res := buildDashboardResponse(api.executor, api.cfg, user)
  262. /*
  263. if len(res.Actions) == 0 {
  264. log.WithFields(log.Fields{
  265. "username": user.Username,
  266. "usergroupLine": user.UsergroupLine,
  267. "provider": user.Provider,
  268. "acls": user.Acls,
  269. "availableActions": len(api.cfg.Actions),
  270. }).Warn("Zero actions found for user")
  271. }
  272. */
  273. log.Tracef("GetDashboardComponents: %v", res)
  274. return connect.NewResponse(res), nil
  275. }
  276. func (api *oliveTinAPI) GetLogs(ctx ctx.Context, req *connect.Request[apiv1.GetLogsRequest]) (*connect.Response[apiv1.GetLogsResponse], error) {
  277. user := acl.UserFromContext(ctx, api.cfg)
  278. ret := &apiv1.GetLogsResponse{}
  279. logEntries, countRemaining := api.executor.GetLogTrackingIds(req.Msg.StartOffset, api.cfg.LogHistoryPageSize)
  280. for _, logEntry := range logEntries {
  281. action := api.cfg.FindAction(logEntry.ActionTitle)
  282. if action == nil || acl.IsAllowedLogs(api.cfg, user, action) {
  283. pbLogEntry := api.internalLogEntryToPb(logEntry, user)
  284. ret.Logs = append(ret.Logs, pbLogEntry)
  285. }
  286. }
  287. ret.CountRemaining = countRemaining
  288. ret.PageSize = api.cfg.LogHistoryPageSize
  289. return connect.NewResponse(ret), nil
  290. }
  291. /*
  292. This function is ONLY a helper for the UI - the arguments are validated properly
  293. on the StartAction -> Executor chain. This is here basically to provide helpful
  294. error messages more quickly before starting the action.
  295. */
  296. func (api *oliveTinAPI) ValidateArgumentType(ctx ctx.Context, req *connect.Request[apiv1.ValidateArgumentTypeRequest]) (*connect.Response[apiv1.ValidateArgumentTypeResponse], error) {
  297. err := executor.TypeSafetyCheck("", req.Msg.Value, req.Msg.Type)
  298. desc := ""
  299. if err != nil {
  300. desc = err.Error()
  301. }
  302. return connect.NewResponse(&apiv1.ValidateArgumentTypeResponse{
  303. Valid: err == nil,
  304. Description: desc,
  305. }), nil
  306. }
  307. func (api *oliveTinAPI) WhoAmI(ctx ctx.Context, req *connect.Request[apiv1.WhoAmIRequest]) (*connect.Response[apiv1.WhoAmIResponse], error) {
  308. user := acl.UserFromContext(ctx, api.cfg)
  309. res := &apiv1.WhoAmIResponse{
  310. AuthenticatedUser: user.Username,
  311. Usergroup: user.UsergroupLine,
  312. Provider: user.Provider,
  313. Sid: user.SID,
  314. Acls: user.Acls,
  315. }
  316. return connect.NewResponse(res), nil
  317. }
  318. func (api *oliveTinAPI) SosReport(ctx ctx.Context, req *connect.Request[apiv1.SosReportRequest]) (*connect.Response[apiv1.SosReportResponse], error) {
  319. sos := installationinfo.GetSosReport()
  320. if !api.cfg.InsecureAllowDumpSos {
  321. log.Info(sos)
  322. sos = "Your SOS Report has been logged to OliveTin logs.\n\nIf you are in a safe network, you can temporarily set `insecureAllowDumpSos: true` in your config.yaml, restart OliveTin, and refresh this page - it will put the output directly in the browser."
  323. }
  324. ret := &apiv1.SosReportResponse{
  325. Alert: sos,
  326. }
  327. return connect.NewResponse(ret), nil
  328. }
  329. func (api *oliveTinAPI) DumpVars(ctx ctx.Context, req *connect.Request[apiv1.DumpVarsRequest]) (*connect.Response[apiv1.DumpVarsResponse], error) {
  330. res := &apiv1.DumpVarsResponse{}
  331. if !api.cfg.InsecureAllowDumpVars {
  332. res.Alert = "Dumping variables is not allowed by default because it is insecure."
  333. return connect.NewResponse(res), nil
  334. }
  335. res.Alert = "Dumping variables has been enabled in the configuration. Please set InsecureAllowDumpVars = false again after you don't need it anymore"
  336. res.Contents = sv.GetAll()
  337. return connect.NewResponse(res), nil
  338. }
  339. func (api *oliveTinAPI) DumpPublicIdActionMap(ctx ctx.Context, req *connect.Request[apiv1.DumpPublicIdActionMapRequest]) (*connect.Response[apiv1.DumpPublicIdActionMapResponse], error) {
  340. res := &apiv1.DumpPublicIdActionMapResponse{}
  341. res.Contents = make(map[string]*apiv1.ActionEntityPair)
  342. if !api.cfg.InsecureAllowDumpActionMap {
  343. res.Alert = "Dumping Public IDs is disallowed."
  344. return connect.NewResponse(res), nil
  345. }
  346. api.executor.MapActionIdToBindingLock.RLock()
  347. for k, v := range api.executor.MapActionIdToBinding {
  348. res.Contents[k] = &apiv1.ActionEntityPair{
  349. ActionTitle: v.Action.Title,
  350. EntityPrefix: v.EntityPrefix,
  351. }
  352. }
  353. api.executor.MapActionIdToBindingLock.RUnlock()
  354. res.Alert = "Dumping variables has been enabled in the configuration. Please set InsecureAllowDumpActionMap = false again after you don't need it anymore"
  355. return connect.NewResponse(res), nil
  356. }
  357. func (api *oliveTinAPI) GetReadyz(ctx ctx.Context, req *connect.Request[apiv1.GetReadyzRequest]) (*connect.Response[apiv1.GetReadyzResponse], error) {
  358. res := &apiv1.GetReadyzResponse{
  359. Status: "OK",
  360. }
  361. return connect.NewResponse(res), nil
  362. }
  363. func (api *oliveTinAPI) EventStream(ctx ctx.Context, req *connect.Request[apiv1.EventStreamRequest], srv *connect.ServerStream[apiv1.EventStreamResponse]) error {
  364. log.Debugf("EventStream: %v", req.Msg)
  365. client := &connectedClients{
  366. channel: make(chan *apiv1.EventStreamResponse, 10), // Buffered channel to hold Events
  367. AuthenticatedUser: acl.UserFromContext(ctx, api.cfg),
  368. }
  369. log.Infof("EventStream: client connected: %v", client.AuthenticatedUser.Username)
  370. api.connectedClients = append(api.connectedClients, client)
  371. // loop over client channel and send events to connectedClient
  372. for msg := range client.channel {
  373. log.Debugf("Sending event to client: %v", msg)
  374. if err := srv.Send(msg); err != nil {
  375. log.Errorf("Error sending event to client: %v", err)
  376. }
  377. }
  378. log.Infof("EventStream: client disconnected")
  379. return nil
  380. }
  381. func (api *oliveTinAPI) OnActionMapRebuilt() {
  382. for _, client := range api.connectedClients {
  383. select {
  384. case client.channel <- &apiv1.EventStreamResponse{
  385. Event: &apiv1.EventStreamResponse_ConfigChanged{
  386. ConfigChanged: &apiv1.EventConfigChanged{},
  387. },
  388. }:
  389. default:
  390. log.Warnf("EventStream: client channel is full, dropping message")
  391. }
  392. }
  393. }
  394. func (api *oliveTinAPI) OnExecutionStarted(ex *executor.InternalLogEntry) {
  395. for _, client := range api.connectedClients {
  396. select {
  397. case client.channel <- &apiv1.EventStreamResponse{
  398. Event: &apiv1.EventStreamResponse_ExecutionStarted{
  399. ExecutionStarted: &apiv1.EventExecutionStarted{
  400. LogEntry: api.internalLogEntryToPb(ex, client.AuthenticatedUser),
  401. },
  402. },
  403. }:
  404. default:
  405. log.Warnf("EventStream: client channel is full, dropping message")
  406. }
  407. }
  408. }
  409. func (api *oliveTinAPI) OnExecutionFinished(ex *executor.InternalLogEntry) {
  410. for _, client := range api.connectedClients {
  411. select {
  412. case client.channel <- &apiv1.EventStreamResponse{
  413. Event: &apiv1.EventStreamResponse_ExecutionFinished{
  414. ExecutionFinished: &apiv1.EventExecutionFinished{
  415. LogEntry: api.internalLogEntryToPb(ex, client.AuthenticatedUser),
  416. },
  417. },
  418. }:
  419. default:
  420. log.Warnf("EventStream: client channel is full, dropping message")
  421. }
  422. }
  423. }
  424. func (api *oliveTinAPI) GetDiagnostics(ctx ctx.Context, req *connect.Request[apiv1.GetDiagnosticsRequest]) (*connect.Response[apiv1.GetDiagnosticsResponse], error) {
  425. res := &apiv1.GetDiagnosticsResponse{
  426. SshFoundKey: installationinfo.Runtime.SshFoundKey,
  427. SshFoundConfig: installationinfo.Runtime.SshFoundConfig,
  428. }
  429. return connect.NewResponse(res), nil
  430. }
  431. func (api *oliveTinAPI) OnOutputChunk(content []byte, executionTrackingId string) {
  432. for _, client := range api.connectedClients {
  433. select {
  434. case client.channel <- &apiv1.EventStreamResponse{
  435. Event: &apiv1.EventStreamResponse_OutputChunk{
  436. OutputChunk: &apiv1.EventOutputChunk{
  437. Output: string(content),
  438. ExecutionTrackingId: executionTrackingId,
  439. },
  440. },
  441. }:
  442. default:
  443. log.Warnf("EventStream: client channel is full, dropping message")
  444. }
  445. }
  446. }
  447. func newServer(ex *executor.Executor) *oliveTinAPI {
  448. server := oliveTinAPI{}
  449. server.cfg = ex.Cfg
  450. server.executor = ex
  451. ex.AddListener(&server)
  452. return &server
  453. }
  454. func GetNewHandler(ex *executor.Executor) (string, http.Handler) {
  455. server := newServer(ex)
  456. return apiv1connect.NewOliveTinApiServiceHandler(server)
  457. }