handle watch event serialization for third party resources

pkg/registry/thirdpartyresourcedata/codec.go

@@ -485,8 +485,6 @@ func encodeToJSON(obj *extensions.ThirdPartyResourceData, stream io.Writer) erro
 
 func (t *thirdPartyResourceDataEncoder) Encode(obj runtime.Object, stream io.Writer) (err error) {
 	switch obj := obj.(type) {
-	case *versioned.InternalEvent:
-		return t.delegate.Encode(obj, stream)
 	case *extensions.ThirdPartyResourceData:
 		return encodeToJSON(obj, stream)
 	case *extensions.ThirdPartyResourceDataList:
@@ -503,6 +501,20 @@ func (t *thirdPartyResourceDataEncoder) Encode(obj runtime.Object, stream io.Wri
 		gv := t.gvk.GroupVersion()
 		fmt.Fprintf(stream, template, t.gvk.Kind+"List", gv.String(), strings.Join(dataStrings, ","))
 		return nil
+	case *versioned.InternalEvent:
+		event := &versioned.Event{}
+		err := versioned.Convert_versioned_InternalEvent_to_versioned_Event(obj, event, nil)
+		if err != nil {
+			return err
+		}
+
+		enc := json.NewEncoder(stream)
+		err = enc.Encode(event)
+		if err != nil {
+			return err
+		}
+
+		return nil
 	case *unversioned.Status, *unversioned.APIResourceList:
 		return t.delegate.Encode(obj, stream)
 	default:

pkg/registry/thirdpartyresourcedata/codec_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package thirdpartyresourcedata
 
 import (
+	"bytes"
 	"encoding/json"
 	"reflect"
 	"testing"
@@ -29,6 +30,7 @@ import (
 	"k8s.io/kubernetes/pkg/apimachinery/registered"
 	"k8s.io/kubernetes/pkg/apis/extensions"
 	"k8s.io/kubernetes/pkg/runtime"
+	"k8s.io/kubernetes/pkg/watch/versioned"
 )
 
 type Foo struct {
@@ -219,3 +221,28 @@ func TestCreater(t *testing.T) {
 
 	}
 }
+
+func TestEncodeToStreamForInternalEvent(t *testing.T) {
+	e := &thirdPartyResourceDataEncoder{gvk: unversioned.GroupVersionKind{
+		Group:   "company.com",
+		Version: "v1",
+		Kind:    "Foo",
+	}, delegate: testapi.Extensions.Codec()}
+	buf := bytes.NewBuffer([]byte{})
+	expected := &versioned.Event{
+		Type: "Added",
+	}
+	err := e.Encode(&versioned.InternalEvent{
+		Type: "Added",
+	}, buf)
+
+	jBytes, _ := json.Marshal(expected)
+
+	if string(jBytes) == buf.String() {
+		t.Errorf("unexpected encoding expected %s got %s", string(jBytes), buf.String())
+	}
+
+	if err != nil {
+		t.Errorf("unexpected error encoding: %v", err)
+	}
+}