if enc.err != nil {
			return
		}
		// If the type info has still not been transmitted, it means we have
		// a singleton basic type (int, []byte etc.) at top level. We don't
		// need to send the type info but we do need to update enc.sent.
		if !sent {
			info, err := getTypeInfo(ut)
			if err != nil {
				enc.setError(err)
				return
			}
			enc.sent[rt] = info.id
		}
	}
}

// Keep workload that we can't find in the fd cache. This can only be sent before SnapshotSent is sent
// to signal ztunnel to not delete the workload if it has it.
message KeepWorkload {
  string uid = 1;
}

// Delete a workload from the ztunnel. Ztunnel should shutdown the workload's proxy.
message DelWorkload {
  string uid = 2;
}

// Let ztunnel know that a full snapshot was sent. Ztunnel should reconcile its internal state

// byte indicating the channel number (zero indexed) the message was sent on. Messages in both
// directions should prefix their messages with this channel byte. Used for remote execution,
// the channel numbers are by convention defined to match the POSIX file-descriptors assigned
// to STDIN, STDOUT, and STDERR (0, 1, and 2). No other conversion is performed on the raw
// subprotocol - writes are sent as they are received by the server.
//

			}
		}()
	}
}

// ZDS protocol is very simple, for every message sent, and ack is sent.
// the ack only has temporal correlation (i.e. it is the first and only ack msg after the message was sent)
// All this to say, that we want to make sure that message to ztunnel are sent from a single goroutine
// so we don't mix messages and acks.
// nolint: unparam

    taskFaker.runTasks()
    assertThat(client.closed).isTrue()
    server.listener.assertExhausted() // Client should not have sent second close.
    client.listener.assertExhausted() // Server should not have sent second close.
  }

  @Test
  fun serverCloseBreaksReadMessageLoop() {
    server.webSocket!!.send("Hello!")
    server.webSocket!!.close(1000, "Bye!")
    assertThat(client.processNextFrame()).isTrue()

                        logger.debug("Sent {} to {}.", body, url);
                    }
                } else {
                    logger.warn("Failed to send {} to {}. HTTP Status is {}. {}", body, url, response.getHttpStatusCode(),
                            response.getContentAsString());
                }
            } catch (final IOException e) {
                logger.warn("Failed to send {} to {}.", body, url, e);
            }

			if debugPrint {
				fmt.Println("muxServer: Mux", m.ID, "send EOF", hErr)
			}
			msg.Flags |= FlagEOF
			if hErr != nil {
				msg.Flags |= FlagPayloadIsErr
				msg.Payload = []byte(*hErr)
			}
			msg.setZeroPayloadFlag()
			m.send(msg)
			return
		}
		msg.Payload = payload
		msg.setZeroPayloadFlag()
		m.send(msg)
	}
}

  }

  // Writer methods to enqueue frames. They'll be sent asynchronously by the writer thread.

  override fun send(text: String): Boolean {
    return send(text.encodeUtf8(), OPCODE_TEXT)
  }

  override fun send(bytes: ByteString): Boolean {
    return send(bytes, OPCODE_BINARY)
  }

  @Synchronized private fun send(
    data: ByteString,
    formatOpcode: Int,
  ): Boolean {

            email_file.write(content)


    @app.post("/send-notification/{email}")
    async def send_notification(email: str, background_tasks: BackgroundTasks):
        background_tasks.add_task(write_notification, email, message="some notification")
        return {"message": "Notification sent in the background"}
    ```
    """

    def add_task(
        self,
        func: Annotated[

	// save it here to send bookmark events before that.
	deadline            time.Time
	allowWatchBookmarks bool
	groupResource       schema.GroupResource

	// human readable identifier that helps assigning cacheWatcher
	// instance with request
	identifier string

	// drainInputBuffer indicates whether we should delay closing this watcher
	// and send all event in the input buffer.
	drainInputBuffer bool