message SnapshotSent {
}

// Ztunnel ack message. If error is not empty, this is an error message.
message Ack {
  string error = 1;
}

/*
Protocol details:
on new connection from ztunnel to CNI, the CNI agent
- will send all the existing payloads (that it has in its cache) to the ztunnel using AddWorkload message.
- the ztunnel will send an ack for each payload (which the CNI will wait for before sending the next one).

}

func sendAck(c *net.UnixConn) {
	ack := &zdsapi.WorkloadResponse{
		Payload: &zdsapi.WorkloadResponse_Ack{
			Ack: &zdsapi.Ack{},
		},
	}
	data, err := proto.Marshal(ack)
	if err != nil {
		panic(err)
	}
	err = c.SetWriteDeadline(time.Now().Add(time.Second))
	if err != nil {
		panic(err)
	}
	c.Write(data)
}

func sendHello(c *net.UnixConn) {
	ack := &zdsapi.ZdsHello{

      .isEqualTo("<< 0x00000000    15 SETTINGS      ")
    assertThat(frameLog(false, 0, 0, TYPE_SETTINGS, FLAG_ACK))
      .isEqualTo(">> 0x00000000     0 SETTINGS      ACK")
    assertThat(frameLog(true, 0, 0, TYPE_SETTINGS, FLAG_ACK))
      .isEqualTo("<< 0x00000000     0 SETTINGS      ACK")
    assertThat(frameLog(true, 3, 22, TYPE_HEADERS, FLAG_END_HEADERS))
      .isEqualTo("<< 0x00000003    22 HEADERS       END_HEADERS")

      this.streamId = streamId
      this.errorCode = errorCode
    }

    override fun ping(
      ack: Boolean,
      payload1: Int,
      payload2: Int,
    ) {
      check(type == -1)
      type = Http2.TYPE_PING
      this.ack = ack
      this.payload1 = payload1
      this.payload2 = payload2
    }

    override fun goAway(
      lastGoodStreamId: Int,

                    project.extensions.create("test", TestExtension::class)
                }
            }

            open class TestExtension {
                fun ack() = println("Ack!")
            }
            """
        )

        withFile(
            "src/test/kotlin/plugin/TestPluginTest.kt",
            """

            package plugin

			response: true,
		},
		{
			name: "ack",
			proxy: &TestProxy{
				WatchedResources: map[string]*WatchedResource{
					model.ClusterType: {
						NonceSent: "nonce",
					},
				},
			},
			request: &discovery.DiscoveryRequest{
				TypeUrl:       model.ClusterType,
				VersionInfo:   "v1",
				ResponseNonce: "nonce",
			},
			response: false,
		},
		{
			name: "ack forced",
			proxy: &TestProxy{

	}
	// Acknowledge Mux created.
	// Send async.
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		var ack message
		ack.Op = OpAckMux
		ack.Flags = m.BaseFlags
		ack.MuxID = m.ID
		m.send(ack)
		if debugPrint {
			fmt.Println("connected stream mux:", ack.MuxID)
		}
	}()

	// Data inbound to the handler
	var handlerIn chan []byte
	if inboundCap > 0 {

	if ok {
		return Key[O](keyFunc(ac.Name, ac.Namespace)), true
	}
	arn, ok := any(a).(ResourceNamer)
	if ok {
		return Key[O](arn.ResourceName()), true
	}
	ack := GetApplyConfigKey(a)
	if ack != nil {
		return *ack, true
	}
	return "", false
}

// getLabels returns the labels for an object, if possible.
// Warning: this will panic if the labels is not available.

		},
	})
	if err != nil {
		t.Fatal(err)
	}

	// Wait until nonce was updated, which represents an ACK/NACK has been received.
	retry.UntilSuccessOrFail(t, func() error {
		if proxy.ecdsLastNonce.Load() == n1.Load() {
			return errors.New("last process nonce has not been updated. no ecds ack/nack is received yet")
		}
		return nil
	}, retry.Timeout(time.Second), retry.Delay(time.Millisecond))

// EventHandler allows for generic monitoring of xDS ACKS and disconnects, for the purpose of tracking
// Config distribution through the mesh.
type DistributionStatusCache interface {
	// RegisterEvent notifies the implementer of an xDS ACK, and must be non-blocking
	RegisterEvent(conID string, eventType EventType, nonce string)
	RegisterDisconnect(s string, types sets.Set[EventType])
	QueryLastNonce(conID string, eventType EventType) (noncePrefix string)