-
Notifications
You must be signed in to change notification settings - Fork 6
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Move Flume message protocol to the flume package
closes #35 This change is moving all message protocol things to the Flume client and now all the code related to the custom clients based on the Avro client (like the Flume client) is encapsulated in the particular package (like the flume package). It should allow building other clients based on the Avro client which may use other messages protocols.
- Loading branch information
Showing
10 changed files
with
301 additions
and
269 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,111 @@ | ||
| package flume | ||
|
|
||
| import ( | ||
| "fmt" | ||
|
|
||
| "github.com/myzhan/avroipc/protocols" | ||
|
|
||
| "github.com/linkedin/goavro" | ||
| ) | ||
|
|
||
| type message struct { | ||
| request *goavro.Codec | ||
| response *goavro.Codec | ||
| errors *goavro.Codec | ||
| } | ||
|
|
||
| // The Avro message protocol implementation for the Avro RPC protocol for using with the Avro Flume Source. | ||
| // | ||
| // It has used for preparing an outgoing message from input data and parsing a response message. | ||
| // | ||
| // The Avro Flume Source haven't documented well now. | ||
| type AvroSourceProtocol struct { | ||
| messages map[string]message | ||
| } | ||
|
|
||
| func NewAvroSource() (protocols.MessageProtocol, error) { | ||
| p := &AvroSourceProtocol{ | ||
| messages: make(map[string]message), | ||
| } | ||
|
|
||
| err := p.init() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return p, nil | ||
| } | ||
|
|
||
| func (p *AvroSourceProtocol) init() (err error) { | ||
| eventCodec, err := goavro.NewCodec(eventSchema) | ||
| if err != nil { | ||
| return | ||
| } | ||
| eventsCodec, err := goavro.NewCodec(eventsSchema) | ||
| if err != nil { | ||
| return | ||
| } | ||
| errorsCodec, err := goavro.NewCodec(errorsSchema) | ||
| if err != nil { | ||
| return | ||
| } | ||
| statusCodec, err := goavro.NewCodec(statusSchema) | ||
| if err != nil { | ||
| return | ||
| } | ||
| p.messages["append"] = message{eventCodec, statusCodec, errorsCodec} | ||
| p.messages["appendBatch"] = message{eventsCodec, statusCodec, errorsCodec} | ||
|
|
||
| return | ||
| } | ||
|
|
||
| func (p *AvroSourceProtocol) PrepareMessage(method string, datum interface{}) ([]byte, error) { | ||
| message, ok := p.messages[method] | ||
| if !ok { | ||
| return nil, fmt.Errorf("unknown method name: %s", method) | ||
| } | ||
|
|
||
| return message.request.BinaryFromNative(nil, datum) | ||
| } | ||
|
|
||
| func (p *AvroSourceProtocol) ParseMessage(method string, responseBytes []byte) (interface{}, []byte, error) { | ||
| message, ok := p.messages[method] | ||
| if !ok { | ||
| return nil, responseBytes, fmt.Errorf("unknown method name: %s", method) | ||
| } | ||
|
|
||
| return message.response.NativeFromBinary(responseBytes) | ||
| } | ||
|
|
||
| func (p *AvroSourceProtocol) ParseError(method string, responseBytes []byte) ([]byte, error) { | ||
| message, ok := p.messages[method] | ||
| if !ok { | ||
| return responseBytes, fmt.Errorf("unknown method name: %s", method) | ||
| } | ||
|
|
||
| response, responseBytes, err := message.errors.NativeFromBinary(responseBytes) | ||
| if err != nil { | ||
| return responseBytes, err | ||
| } | ||
|
|
||
| responseMap, ok := response.(map[string]interface{}) | ||
| if !ok { | ||
| return responseBytes, fmt.Errorf("cannot convert error union to map: %v", response) | ||
| } | ||
|
|
||
| responseInt, ok := responseMap["string"] | ||
| if !ok { | ||
| return responseBytes, fmt.Errorf("string error not found in map: %v", responseMap) | ||
| } | ||
|
|
||
| responseStr, ok := responseInt.(string) | ||
| if !ok { | ||
| return responseBytes, fmt.Errorf("cannot convert string error to string: %v", responseInt) | ||
| } | ||
|
|
||
| return responseBytes, fmt.Errorf(responseStr) | ||
| } | ||
|
|
||
| func (p *AvroSourceProtocol) GetSchema() string { | ||
| return messageProtocol | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,123 @@ | ||
| package flume | ||
|
|
||
| const errorsSchema = ` | ||
| { | ||
| "type": [ | ||
| "string" | ||
| ] | ||
| } | ||
| ` | ||
|
|
||
| const eventSchema = ` | ||
| { | ||
| "type": "record", | ||
| "name": "AvroFlumeEvent", | ||
| "fields": [ | ||
| { | ||
| "name": "headers", | ||
| "type": { | ||
| "type": "map", | ||
| "values": "string" | ||
| } | ||
| }, | ||
| { | ||
| "name": "body", | ||
| "type": "bytes" | ||
| } | ||
| ] | ||
| } | ||
| ` | ||
|
|
||
| const eventsSchema = ` | ||
| { | ||
| "type": "array", | ||
| "items": { | ||
| "type": "record", | ||
| "name": "AvroFlumeEvent", | ||
| "fields": [ | ||
| { | ||
| "name": "headers", | ||
| "type": { | ||
| "type": "map", | ||
| "values": "string" | ||
| } | ||
| }, | ||
| { | ||
| "name": "body", | ||
| "type": "bytes" | ||
| } | ||
| ] | ||
| } | ||
| } | ||
| ` | ||
|
|
||
| const statusSchema = ` | ||
| { | ||
| "type": "enum", | ||
| "name": "Status", | ||
| "symbols": [ | ||
| "OK", | ||
| "FAILED", | ||
| "UNKNOWN" | ||
| ] | ||
| } | ||
| ` | ||
|
|
||
| const messageProtocol = ` | ||
| { | ||
| "protocol": "AvroSourceProtocol", | ||
| "namespace": "org.apache.flume.source.avro", | ||
| "doc": "* Licensed to the Apache Software Foundation (ASF) under one\n * or more contributor license agreements. See the NOTICE file\n * distributed with this work for additional information\n * regarding copyright ownership. The ASF licenses this file\n * to you under the Apache License, Version 2.0 (the\n * \"License\"); you may not use this file except in compliance\n * with the License. You may obtain a copy of the License at\n *\n * http:https://www.apache.org/licenses/LICENSE-2.0\n *\n * Unless required by applicable law or agreed to in writing,\n * software distributed under the License is distributed on an\n * \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n * KIND, either express or implied. See the License for the\n * specific language governing permissions and limitations\n * under the License.", | ||
| "types": [ | ||
| { | ||
| "type": "enum", | ||
| "name": "Status", | ||
| "symbols": [ | ||
| "OK", | ||
| "FAILED", | ||
| "UNKNOWN" | ||
| ] | ||
| }, | ||
| { | ||
| "type": "record", | ||
| "name": "AvroFlumeEvent", | ||
| "fields": [ | ||
| { | ||
| "name": "headers", | ||
| "type": { | ||
| "type": "map", | ||
| "values": "string" | ||
| } | ||
| }, | ||
| { | ||
| "name": "body", | ||
| "type": "bytes" | ||
| } | ||
| ] | ||
| } | ||
| ], | ||
| "messages": { | ||
| "append": { | ||
| "request": [ | ||
| { | ||
| "name": "event", | ||
| "type": "AvroFlumeEvent" | ||
| } | ||
| ], | ||
| "response": "Status" | ||
| }, | ||
| "appendBatch": { | ||
| "request": [ | ||
| { | ||
| "name": "events", | ||
| "type": { | ||
| "type": "array", | ||
| "items": "AvroFlumeEvent" | ||
| } | ||
| } | ||
| ], | ||
| "response": "Status" | ||
| } | ||
| } | ||
| } | ||
| ` |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.