Move CloudEvent from EventGrid to to azure-core (Azure#19193)

Author: YijunXieMS

eng/versioning/version_client.txt

@@ -218,6 +218,8 @@ unreleased_com.azure:azure-messaging-eventhubs;5.5.0
 unreleased_com.azure:azure-messaging-eventhubs-checkpointstore-blob;1.5.0
 unreleased_com.azure:azure-core-management;1.2.0-beta.1
 unreleased_com.azure:azure-core-tracing-opentelemetry;1.0.0-beta.8
+unreleased_com.azure:azure-core;1.14.0-beta.1
+unreleased_com.azure:azure-core-serializer-json-jackson;1.2.0-beta.1
 
 # Released Beta dependencies: Copy the entry from above, prepend "beta_", remove the current
 # version and set the version to the released beta. Released beta dependencies are only valid

sdk/core/azure-core/src/main/java/com/azure/core/implementation/serializer/JacksonSerializer.java

@@ -0,0 +1,54 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.core.implementation.serializer;
+
+import com.azure.core.util.logging.ClientLogger;
+import com.azure.core.util.serializer.JacksonAdapter;
+import com.azure.core.util.serializer.JsonSerializer;
+import com.azure.core.util.serializer.SerializerAdapter;
+import com.azure.core.util.serializer.SerializerEncoding;
+import com.azure.core.util.serializer.TypeReference;
+import reactor.core.publisher.Mono;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.UncheckedIOException;
+
+/**
+ * Jackson based implementation of the {@link JsonSerializer}. This can be used as a default Json serializer if
+ * no JsonSerializerProvider is in the class path.
+ */
+public class JacksonSerializer implements JsonSerializer {
+    private final SerializerAdapter jacksonAdapter = JacksonAdapter.createDefaultSerializerAdapter();
+    private final ClientLogger logger = new ClientLogger(JacksonSerializer.class);
+
+    @Override
+    public <T> T deserialize(InputStream stream, TypeReference<T> typeReference) {
+        try {
+            return jacksonAdapter.deserialize(stream, typeReference.getJavaType(), SerializerEncoding.JSON);
+        } catch (IOException e) {
+            throw logger.logExceptionAsError(new UncheckedIOException(e));
+        }
+    }
+
+    @Override
+    public <T> Mono<T> deserializeAsync(InputStream stream, TypeReference<T> typeReference) {
+        return Mono.defer(() -> Mono.fromCallable(() -> deserialize(stream, typeReference)));
+    }
+
+    @Override
+    public void serialize(OutputStream stream, Object value) {
+        try {
+            jacksonAdapter.serialize(value, SerializerEncoding.JSON, stream);
+        } catch (IOException e) {
+            throw logger.logExceptionAsError(new UncheckedIOException(e));
+        }
+    }
+
+    @Override
+    public Mono<Void> serializeAsync(OutputStream stream, Object value) {
+        return Mono.fromRunnable(() -> serialize(stream, value));
+    }
+}